Serve MCP over the process's stdin and stdout.
While serving, fd 0 points at the null device and fd 1 at stderr, so handlers
and children read EOF and their stray output misses the wire; both descriptors
are restored on exit. Explicit streams skip the claim, and a second concurrent
stdio_server() raises RuntimeError.
Source code in src/mcp/server/stdio.py
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170 | @asynccontextmanager
async def stdio_server(stdin: anyio.AsyncFile[str] | None = None, stdout: anyio.AsyncFile[str] | None = None):
"""Serve MCP over the process's stdin and stdout.
While serving, fd 0 points at the null device and fd 1 at stderr, so handlers
and children read EOF and their stray output misses the wire; both descriptors
are restored on exit. Explicit streams skip the claim, and a second concurrent
stdio_server() raises RuntimeError.
"""
# Re-wrap the binary buffers as UTF-8 text; the std handles' platform encodings are unreliable.
restore_stdin: Callable[[], None] | None = None
restore_stdout: Callable[[], None] | None = None
try:
if not stdin:
stdin_buffer, restore_stdin = _claim_fd(0, sys.stdin, "rb", _open_stdin_diversion)
stdin = anyio.wrap_file(TextIOWrapper(stdin_buffer, encoding="utf-8", errors="replace"))
if not stdout:
stdout_buffer, restore_stdout = _claim_fd(1, sys.stdout, "wb", _open_stdout_diversion)
stdout = anyio.wrap_file(TextIOWrapper(stdout_buffer, encoding="utf-8"))
read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](0)
write_stream, write_stream_reader = create_context_streams[SessionMessage](0)
async def stdin_reader():
try:
async with read_stream_writer:
async for line in stdin:
try:
message = types.jsonrpc_message_adapter.validate_json(line, by_name=False)
except Exception as exc:
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_unset=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
finally:
if restore_stdout is not None:
restore_stdout()
if restore_stdin is not None:
restore_stdin()
|