content: """
  Stdio Server Transport Module
  
  This module provides functionality for creating an stdio-based transport layer
  that can be used to communicate with an MCP client through standard input/output
  streams.
  
  Example usage:
  ```
      async def run_server():
          async with stdio_server() as (read_stream, write_stream):
              # read_stream contains incoming JSONRPCMessages from stdin
              # write_stream allows sending JSONRPCMessages to stdout
              server = await create_my_server()
              await server.run(read_stream, write_stream, init_options)
  
      anyio.run(run_server)
  ```
  """
  
  import sys
  from contextlib import asynccontextmanager
  from io import TextIOWrapper
  
  import anyio
  import anyio.lowlevel
  from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
  
  import mcp.types as types
  from mcp.shared.message import SessionMessage
  
  
  @asynccontextmanager
  async def stdio_server(
      stdin: anyio.AsyncFile[str] | None = None,
      stdout: anyio.AsyncFile[str] | None = None,
  ):
      """
      Server transport for stdio: this communicates with an MCP client by reading
      from the current process' stdin and writing to stdout.
      """
      # Purposely not using context managers for these, as we don't want to close
      # standard process handles. Encoding of stdin/stdout as text streams on
      # python is platform-dependent (Windows is particularly problematic), so we
      # re-wrap the underlying binary stream to ensure UTF-8.
      if not stdin:
          stdin = anyio.wrap_file(TextIOWrapper(sys.stdin.buffer, encoding="utf-8"))
      if not stdout:
          stdout = anyio.wrap_file(TextIOWrapper(sys.stdout.buffer, encoding="utf-8"))
  
      read_stream: MemoryObjectReceiveStream[SessionMessage | Exception]
      read_stream_writer: MemoryObjectSendStream[SessionMessage | Exception]
  
      write_stream: MemoryObjectSendStream[SessionMessage]
      write_stream_reader: MemoryObjectReceiveStream[SessionMessage]
  
      read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
      write_stream, write_stream_reader = anyio.create_memory_object_stream(0)
  
      async def stdin_reader():
          try:
              async with read_stream_writer:
                  async for line in stdin:
                      try:
                          message = types.JSONRPCMessage.model_validate_json(line)
                      except Exception as exc:  # pragma: no cover
                          await read_stream_writer.send(exc)
                          continue
  
                      session_message = SessionMessage(message)
                      await read_stream_writer.send(session_message)
          except anyio.ClosedResourceError:  # pragma: no cover
              await anyio.lowlevel.checkpoint()
  
      async def stdout_writer():
          try:
              async with write_stream_reader:
                  async for session_message in write_stream_reader:
                      json = session_message.message.model_dump_json(by_alias=True, exclude_none=True)
                      await stdout.write(json + "\n")
                      await stdout.flush()
          except anyio.ClosedResourceError:  # pragma: no cover
              await anyio.lowlevel.checkpoint()
  
      async with anyio.create_task_group() as tg:
          tg.start_soon(stdin_reader)
          tg.start_soon(stdout_writer)
          yield read_stream, write_stream
  