Function post_writer

src/mcp/client/sse.py:118–143

(endpoint_url: str)

                    await read_stream_writer.aclose()

            async def post_writer(endpoint_url: str):
                try:
                    async with write_stream_reader, write_stream:

                        async def _send_message(session_message: SessionMessage) -> None:
                            logger.debug(f"Sending client message: {session_message}")
                            response = await client.post(
                                endpoint_url,
                                json=session_message.message.model_dump(
                                    by_alias=True,
                                    mode="json",
                                    exclude_unset=True,
                                ),
                            )
                            response.raise_for_status()
                            logger.debug(f"Client message sent successfully: {response.status_code}")

                        async for session_message in write_stream_reader:
                            sender_ctx = write_stream_reader.last_context
                            if sender_ctx is not None:
                                async with anyio.create_task_group() as tg:
                                    sender_ctx.run(tg.start_soon, _send_message, session_message)
                            else:
                                await _send_message(session_message)  # pragma: no cover
                except Exception:  # pragma: lax no cover
                    logger.exception("Error in post_writer")

            # On Python 3.14, coverage.py reports a phantom branch arc on this
            # line (->yield) when nested two async-with levels deep. The branch