modelcontextprotocol
diff --git a/src/mcp/shared/session.py b/src/mcp/shared/session.py
@@ -455,6 +455,27 @@ async def _receive_loop(self) -> None:
                         pass
                 self._response_streams.clear()
 
+    def _normalize_request_id(self, response_id: RequestId) -> RequestId:
+        """
+        Normalize a response ID to match how request IDs are stored.
+
+        Since the client always sends integer IDs, we normalize string IDs
+        to integers when possible. This matches the TypeScript SDK approach:
+        https://github.com/modelcontextprotocol/typescript-sdk/blob/a606fb17909ea454e83aab14c73f14ea45c04448/src/shared/protocol.ts#L861
+
+        Args:
+            response_id: The response ID from the incoming message.
+
+        Returns:
+            The normalized ID (int if possible, otherwise original value).
+        """
+        if isinstance(response_id, str):
+            try:
+                return int(response_id)
+            except ValueError:
+                logging.warning(f"Response ID {response_id!r} cannot be normalized to match pending requests")
+        return response_id
+
     async def _handle_response(self, message: SessionMessage) -> None:
         """
         Handle an incoming response or error message.
@@ -471,7 +492,8 @@ async def _handle_response(self, message: SessionMessage) -> None:
         if not isinstance(root, JSONRPCResponse | JSONRPCError):
             return  # pragma: no cover
 
-        response_id: RequestId = root.id
+        # Normalize response ID to handle type mismatches (e.g., "0" vs 0)
+        response_id = self._normalize_request_id(root.id)
 
         # First, check response routers (e.g., TaskResultHandler)
         if isinstance(root, JSONRPCError):

diff --git a/tests/shared/test_session.py b/tests/shared/test_session.py
@@ -9,12 +9,18 @@
 from mcp.server.lowlevel.server import Server
 from mcp.shared.exceptions import McpError
 from mcp.shared.memory import create_client_server_memory_streams, create_connected_server_and_client_session
+from mcp.shared.message import SessionMessage
 from mcp.types import (
     CancelledNotification,
     CancelledNotificationParams,
     ClientNotification,
     ClientRequest,
     EmptyResult,
+    ErrorData,
+    JSONRPCError,
+    JSONRPCMessage,
+    JSONRPCRequest,
+    JSONRPCResponse,
     TextContent,
 )
 
@@ -122,6 +128,171 @@ async def make_request(client_session: ClientSession):
                 await ev_cancelled.wait()
 
 
+@pytest.mark.anyio
+async def test_response_id_type_mismatch_string_to_int():
+    """
+    Test that responses with string IDs are correctly matched to requests sent with
+    integer IDs.
+
+    This handles the case where a server returns "id": "0" (string) but the client
+    sent "id": 0 (integer). Without ID type normalization, this would cause a timeout.
+    """
+    ev_response_received = anyio.Event()
+    result_holder: list[types.EmptyResult] = []
+
+    async with create_client_server_memory_streams() as (client_streams, server_streams):
+        client_read, client_write = client_streams
+        server_read, server_write = server_streams
+
+        async def mock_server():
+            """Receive a request and respond with a string ID instead of integer."""
+            message = await server_read.receive()
+            assert isinstance(message, SessionMessage)
+            root = message.message.root
+            assert isinstance(root, JSONRPCRequest)
+            # Get the original request ID (which is an integer)
+            request_id = root.id
+            assert isinstance(request_id, int), f"Expected int, got {type(request_id)}"
+
+            # Respond with the ID as a string (simulating a buggy server)
+            response = JSONRPCResponse(
+                jsonrpc="2.0",
+                id=str(request_id),  # Convert to string to simulate mismatch
+                result={},
+            )
+            await server_write.send(SessionMessage(message=JSONRPCMessage(response)))
+
+        async def make_request(client_session: ClientSession):
+            nonlocal result_holder
+            # Send a ping request (uses integer ID internally)
+            result = await client_session.send_ping()
+            result_holder.append(result)
+            ev_response_received.set()
+
+        async with (
+            anyio.create_task_group() as tg,
+            ClientSession(read_stream=client_read, write_stream=client_write) as client_session,
+        ):
+            tg.start_soon(mock_server)
+            tg.start_soon(make_request, client_session)
+
+            with anyio.fail_after(2):
+                await ev_response_received.wait()
+
+    assert len(result_holder) == 1
+    assert isinstance(result_holder[0], EmptyResult)
+
+
+@pytest.mark.anyio
+async def test_error_response_id_type_mismatch_string_to_int():
+    """
+    Test that error responses with string IDs are correctly matched to requests
+    sent with integer IDs.
+
+    This handles the case where a server returns an error with "id": "0" (string)
+    but the client sent "id": 0 (integer).
+    """
+    ev_error_received = anyio.Event()
+    error_holder: list[McpError] = []
+
+    async with create_client_server_memory_streams() as (client_streams, server_streams):
+        client_read, client_write = client_streams
+        server_read, server_write = server_streams
+
+        async def mock_server():
+            """Receive a request and respond with an error using a string ID."""
+            message = await server_read.receive()
+            assert isinstance(message, SessionMessage)
+            root = message.message.root
+            assert isinstance(root, JSONRPCRequest)
+            request_id = root.id
+            assert isinstance(request_id, int)
+
+            # Respond with an error, using the ID as a string
+            error_response = JSONRPCError(
+                jsonrpc="2.0",
+                id=str(request_id),  # Convert to string to simulate mismatch
+                error=ErrorData(code=-32600, message="Test error"),
+            )
+            await server_write.send(SessionMessage(message=JSONRPCMessage(error_response)))
+
+        async def make_request(client_session: ClientSession):
+            nonlocal error_holder
+            try:
+                await client_session.send_ping()
+                pytest.fail("Expected McpError to be raised")  # pragma: no cover
+            except McpError as e:
+                error_holder.append(e)
+                ev_error_received.set()
+
+        async with (
+            anyio.create_task_group() as tg,
+            ClientSession(read_stream=client_read, write_stream=client_write) as client_session,
+        ):
+            tg.start_soon(mock_server)
+            tg.start_soon(make_request, client_session)
+
+            with anyio.fail_after(2):
+                await ev_error_received.wait()
+
+    assert len(error_holder) == 1
+    assert "Test error" in str(error_holder[0])
+
+
+@pytest.mark.anyio
+async def test_response_id_non_numeric_string_no_match():
+    """
+    Test that responses with non-numeric string IDs don't incorrectly match
+    integer request IDs.
+
+    If a server returns "id": "abc" (non-numeric string), it should not match
+    a request sent with "id": 0 (integer).
+    """
+    ev_timeout = anyio.Event()
+
+    async with create_client_server_memory_streams() as (client_streams, server_streams):
+        client_read, client_write = client_streams
+        server_read, server_write = server_streams
+
+        async def mock_server():
+            """Receive a request and respond with a non-numeric string ID."""
+            message = await server_read.receive()
+            assert isinstance(message, SessionMessage)
+
+            # Respond with a non-numeric string ID (should not match)
+            response = JSONRPCResponse(
+                jsonrpc="2.0",
+                id="not_a_number",  # Non-numeric string
+                result={},
+            )
+            await server_write.send(SessionMessage(message=JSONRPCMessage(response)))
+
+        async def make_request(client_session: ClientSession):
+            try:
+                # Use a short timeout since we expect this to fail
+                from datetime import timedelta
+
+                await client_session.send_request(
+                    ClientRequest(types.PingRequest()),
+                    types.EmptyResult,
+                    request_read_timeout_seconds=timedelta(seconds=0.5),
+                )
+                pytest.fail("Expected timeout")  # pragma: no cover
+            except McpError as e:
+                assert "Timed out" in str(e)
+                ev_timeout.set()
+
+        async with (
+            anyio.create_task_group() as tg,
+            ClientSession(read_stream=client_read, write_stream=client_write) as client_session,
+        ):
+            tg.start_soon(mock_server)
+            tg.start_soon(make_request, client_session)
+
+            with anyio.fail_after(2):
+                await ev_timeout.wait()
+
+
 @pytest.mark.anyio
 async def test_connection_closed():
     """