This PR fixes a nasty but in the way we handle incoming messages, which currently leads to hanging connections when the clients sends notifications.

The observed issue is as follows:

• A user creates a ClientSession and connects to a server.
• Server sends messages such as logging events.
• Upon then next request from the client, the client await hangs.

When I looked into this, I suspected that this must be some channel issues. We use in memory channels to handle messages and it has historically caused issues. In particular, we use channels with a max_lenght of 0, effectively making them non buffered. This is one, if not the only part of the codebase that can block as observed.

So I looked into the two parts where we use channels, once for read and write stream from the sse/stdio implementation to the ClientSession and then within the shared BaseSession class to handle messages separately in the server and client session.

I first changed in client/stdio.py the lines:

    read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
    write_stream, write_stream_reader = anyio.create_memory_object_stream(0)

to

    read_stream_writer, read_stream = anyio.create_memory_object_stream(math.inf)
    write_stream, write_stream_reader = anyio.create_memory_object_stream(math.inf)

and tested again. This fixed the bug! However clearly we are now in bounded memory stream territory which we cannot do without growing buffers in memory linearly. I tested da bit more and boiled it down to

   read_stream_writer, read_stream = anyio.create_memory_object_stream(math.inf)

fixing the issue. So we know something is funky in the read stream.

I looked into @sheffler's issue #201 and PR #202 and confirmed that removing in src/mcp/shared/session.py

                            await self._incoming_message_stream_writer.send(
                                notification
                            )

will fix the probelm as well. So what does it tell us? It tells us that is that await self._incoming_message_stream_writer is blocking. Which in turn tells us that nobody ever reads from self._incoming_message_stream_reader. So where is self._incoming_message_stream_reader used? In BaseSession.incoming_messages.

Now we have theory:

1. incoming_messages is used to read incoming messages on the server.
2. But it's in the BaseSession, so it means the client also offers it.
3. Most clients don't read client.incoming_messages.
4. IF clients don't read client.incoming_messages, the unbuffered stream never empties and once one object is placed, the next write will block.

Now that we have the culprit we can figure out what to do. In my mind the probelm happens because we can't guarantee a consumer of incoming messages, however some implementations might want it. Similarly on the server side, incoming_messages is always used, since ServerSession is used inside Server.

Okay so the goal was then to:

• Make consumption of incoming messages optionally for clients
• Make consumption of incoming message required for servers.

So on the server side incoming_messages is fine, but the client needs a way to react if the developer is interested in. This means the approach of providing it via incoming_messages doesnt work. After thinking of ways to automatically drain, I came up what I think is best:

1. The client offers a callback for all messages. This fits into the client interface using callbacks already.
2. The server continues to expose incoming_messages.
3. The BaseSession offers an abstraction for ClientSession and ServerSession to observe events from the _receive_loop but determine what to do by themselve.