From 8760a1c5be317fec9526f5a61d5cec14505e66a1 Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Sat, 26 Sep 2026 09:17:46 -0600 Subject: [PATCH] Add owned stdio clients for local MCP servers --- ROADMAP.md | 15 +- docs/make.jl | 1 + docs/src/api.md | 8 + docs/src/index.md | 11 +- docs/src/stdio.md | 109 ++++++++ src/ModelContextProtocol.jl | 1 + src/client.jl | 27 +- src/jsonrpc.jl | 8 +- src/stdio.jl | 453 +++++++++++++++++++++++++++++++++ src/types.jl | 31 ++- test/runtests.jl | 1 + test/stdio.jl | 486 ++++++++++++++++++++++++++++++++++++ test/stdio_peer.jl | 116 +++++++++ 13 files changed, 1256 insertions(+), 11 deletions(-) create mode 100644 docs/src/stdio.md create mode 100644 src/stdio.jl create mode 100644 test/stdio.jl create mode 100644 test/stdio_peer.jl diff --git a/ROADMAP.md b/ROADMAP.md index e69c147..2baba61 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -19,6 +19,11 @@ The package supports Streamable HTTP for two protocol eras: The general server accepts both versions by default. The client defaults to `2025-11-25`. Applications opt in to `2026-07-28` with `MCPClientConfig`. +The client also owns local child processes through +`ModelContextProtocol.prepare_stdio_client`. It supports newline-delimited +JSON-RPC, concurrent calls, notifications, legacy server requests, cancellation, +and bounded process shutdown for either explicitly selected protocol version. + The package also includes a separate tools-only server for JuliaC `--trim=safe` builds. That server intentionally supports only the documented `2025-11-25` subset. @@ -31,6 +36,8 @@ The repository tests these areas: Windows. - OAuth 2 and OAuth 3 compatibility. - Stateful and stateless HTTP client/server integration. +- Owned stdio child processes, concurrent response correlation, malformed or + oversized frames, cancellation, blocked callbacks, and process shutdown. - Strict JSON-RPC parsing and notification side-effect rules. - Client result response IDs must match the request before results or session state are accepted. @@ -42,7 +49,12 @@ The repository tests these areas: ## Intentional limits -- The transport is HTTP only. The package does not provide a stdio transport. +- Stdio is a client transport only. It does not provide a stdio server, + protocol auto-detection, automatic restart/replay, modern subscriptions, + HTTP headers/OAuth, or automatic Agentif tool-catalog import. Callbacks must + cooperate with shutdown; a blocked callback produces an explicit close error. +- The general client, including subprocess stdio, is not a JuliaC trim-safe + API. The separate static server remains the supported native subset. - Tool input and output schemas are advertised but are not a complete runtime JSON Schema validation engine. A handler must still validate domain rules. - Modern request-scoped progress and log events keep their correct order, but @@ -63,7 +75,6 @@ The repository tests these areas: without ending the server. 4. Evaluate a lightweight JSON Schema validator for tool arguments and structured results. -5. Add a stdio transport only if a concrete Julia deployment needs it. Do not add a feature only to increase surface coverage. Preserve the small export surface. Keep specialized helpers under the `ModelContextProtocol` diff --git a/docs/make.jl b/docs/make.jl index 52b78b7..bf8ba30 100644 --- a/docs/make.jl +++ b/docs/make.jl @@ -18,6 +18,7 @@ makedocs( ), pages=[ "Home" => "index.md", + "Local stdio client" => "stdio.md", "MCP 2026-07-28" => "protocol-2026.md", "Auth0 Federation Example" => "auth0.md", "Trim-safe static server" => "static-server.md", diff --git a/docs/src/api.md b/docs/src/api.md index 22c9256..fe3f926 100644 --- a/docs/src/api.md +++ b/docs/src/api.md @@ -75,6 +75,7 @@ for the full guide. ## Client Functions +- `ModelContextProtocol.prepare_stdio_client` - `discover_server` - `prepare_manual_client` - `attach_token!` @@ -98,6 +99,13 @@ for the full guide. - `stop_event_listener!` - `terminate_session!` +### Owned stdio clients + +```@docs +ModelContextProtocol.prepare_stdio_client +Base.close(::MCPClient) +``` + ## MCP 2026-07-28 The modern low-level helpers stay under the package namespace to keep the diff --git a/docs/src/index.md b/docs/src/index.md index 8221187..ca766be 100644 --- a/docs/src/index.md +++ b/docs/src/index.md @@ -1,10 +1,10 @@ # ModelContextProtocol.jl `ModelContextProtocol.jl` provides Julia server and client utilities for the -Model Context Protocol (MCP). It focuses on Streamable HTTP servers, discovery -manifests, OAuth-protected resources, JSON-RPC request handling, tools, prompts, -resources, completions, logging notifications, and lightweight client smoke -tests. +Model Context Protocol (MCP). It supports Streamable HTTP servers and clients, +local subprocess clients, discovery manifests, OAuth-protected resources, +JSON-RPC request handling, tools, prompts, resources, completions, and logging +notifications. The package supports the stateful MCP `2025-11-25` protocol and the stateless MCP `2026-07-28` protocol over Streamable HTTP. Clients default to @@ -12,6 +12,9 @@ MCP `2026-07-28` protocol over Streamable HTTP. Clients default to modern client setup, capability checks, multi-round-trip results, custom headers, and subscriptions. +For a server launched as a local command, see the [stdio client guide](stdio.md). +The same initialization, list, and call APIs work with an owned child process. + For deployments that require a concrete request graph, see the [trim-safe static tools server](static-server.md). This API stays under the `ModelContextProtocol` namespace because it is a specialized alternative to diff --git a/docs/src/stdio.md b/docs/src/stdio.md new file mode 100644 index 0000000..e8368c9 --- /dev/null +++ b/docs/src/stdio.md @@ -0,0 +1,109 @@ +# Local stdio client + +Use `ModelContextProtocol.prepare_stdio_client(command)` to launch a local MCP +server and own its stdin/stdout connection. The child must exchange one UTF-8 +JSON-RPC object per line. Diagnostic output belongs on stderr. The constructor +accepts a Julia `Cmd`; it does not run a shell or interpret a command string. + +This example uses the repository's deterministic echo peer. Replace `command` +with your server's command in an application. The do-block closes the child +when the body returns or throws. + +```@example stdio +using ModelContextProtocol + +peer = joinpath(pkgdir(ModelContextProtocol), "test", "stdio_peer.jl") +project = dirname(Base.active_project()) +command = `$(Base.julia_cmd()) --startup-file=no --project=$project $peer` + +ModelContextProtocol.prepare_stdio_client(command; stderr=devnull) do client + initialize_client!(client) + @assert list_tools(client)["tools"][1]["name"] == "echo" + result = call_tool(client, "echo"; arguments=Dict("message" => "Hello, λ")) + println(result["structuredContent"]["message"]) +end +``` + +Without a do-block, use `try`/`finally` and call `close(client)` or +`terminate_session!(client)`. The client owns the direct child and its protocol +pipes. It does not manage a process tree created by that child. Pass the server +executable directly when possible. + +## Protocol versions + +The default is MCP `2025-11-25`. Select `2026-07-28` explicitly with +`MCPClientConfig(protocol_version=ModelContextProtocol.PROTOCOL_VERSION_2026_07_28)`. + +| Behavior | 2025-11-25 | 2026-07-28 | +|:--|:--|:--| +| `initialize_client!` | Initialize, then initialized notification | `server/discover` | +| Request identity/capabilities | Initialization parameters | Per-request `_meta` | +| Lists, calls, and notifications | Supported | Supported | +| Server requests | Existing registered request handlers | Rejected by the protocol | +| Multi-round-trip input | Application-managed | Existing `input_required` helpers | +| Request timeout | Cancellation notification | Cancellation notification | +| Subscriptions | Existing legacy resource calls | Not implemented | + +The implementation follows the dated +[2025-11-25 transport](https://modelcontextprotocol.io/specification/2025-11-25/basic/transports) +and [lifecycle](https://modelcontextprotocol.io/specification/2025-11-25/basic/lifecycle) +rules, and the +[2026-07-28 stdio transport](https://modelcontextprotocol.io/specification/2026-07-28/basic/transports/stdio) +rules for the operations above. It does not probe versions, fall back to another +protocol, restart a child, or replay requests automatically. HTTP headers, +OAuth bearer tokens, HTTP event streams, and modern `subscriptions/listen` are +rejected, as are custom HTTP adapters and verbose HTTP logging. Stdio has no +HTTP header transport; `x-mcp-header` tool arguments are +sent as ordinary arguments. Importing a server's catalog into Agentif is a +separate application concern. + +## Concurrent calls and handlers + +Register notification/request handlers and complete initialization before +starting concurrent calls. Calls receive unique string IDs; out-of-order +responses are matched by the exact ID. A late response to an expired call is +discarded. EOF, invalid JSON/UTF-8, mismatched IDs, and oversized frames fail +pending calls and start process cleanup. + +Notifications and legacy server requests use the existing handler registration +APIs. One callback task preserves arrival order, independently of response +reading. A handler can make a nested client call. A slow handler delays other +callbacks but does not stop response routing. Closing discards queued callbacks +that have not started. + +## Bounds and shutdown + +`config.timeout.readtimeout` defaults to 120 seconds. Each call can override it +with a positive integer `timeout_ms`. This deadline covers queued writes and +response waits. It does not preempt application JSON serialization or OS process +creation. A response timeout sends cancellation, except for legacy initialize; +a write timeout closes the connection because a partial frame cannot be safely +replayed. `connecttimeout` has no effect for a local process; other timeout +settings are rejected. + +`max_message_bytes` defaults to 16 MiB per incoming/outgoing message. +`max_pending_messages` defaults to 128 and separately limits pending calls, +queued writes, and queued callbacks. A full call/write queue reports +`MCPError(:transport_busy)`. Callback overflow fails the connection with +`MCPError(:callback_overflow)`. These are queue/frame limits, not a total memory +quota for parsed JSON or user code. + +Stderr can go directly to a filename, open file, terminal, pipe, or `devnull`. +Caller-provided destinations remain caller-owned. In-memory/custom IO sinks +are rejected because their implicit copy tasks cannot be bounded by the client. +If a pipe destination blocks, requests still have deadlines and closing can +terminate the child. + +`close(client; timeout=5.0)` stops new calls, fails pending calls, closes stdin, +and escalates to process termination and kill if necessary. It waits for owned +IO and callback tasks within the supplied deadline. Cleanup state remains +available if close times out, so a later `close` can finish waiting. + +Julia cannot safely interrupt arbitrary callback code. A blocked user callback +can produce `MCPError(:callback_timeout)` after process/IO cleanup. Release that +callback and close again. A callback may call `close` itself; close skips waiting +for that callback, which finishes when its handler returns. This transport does +not provide forced task cancellation. + +The dynamic subprocess client is not a JuliaC `--trim=safe` API. The package's +[static tools server](static-server.md) remains its supported native subset. diff --git a/src/ModelContextProtocol.jl b/src/ModelContextProtocol.jl index 1cf00fd..e0afde9 100644 --- a/src/ModelContextProtocol.jl +++ b/src/ModelContextProtocol.jl @@ -14,6 +14,7 @@ include("jsonrpc.jl") include("server.jl") include("static_server.jl") include("client.jl") +include("stdio.jl") include("apps.jl") export MCPError, MCPAuthenticationRequired diff --git a/src/client.jl b/src/client.jl index a0aa539..41476ed 100644 --- a/src/client.jl +++ b/src/client.jl @@ -63,7 +63,7 @@ default_client_info() = Dict( function list_tools(client::MCPClient; cursor=nothing, limit=nothing, headers=nothing, timeout_ms=nothing) result = list_entities(client, JSONRPC_METHOD_TOOLS_LIST; cursor=cursor, limit=limit, headers=headers, timeout_ms=timeout_ms) - client_is_modern(client) || return result + (client_is_modern(client) && client.transport.kind == :http) || return result result isa AbstractDict || return result tools = get(result, "tools", nothing) tools isa AbstractVector || return result @@ -118,6 +118,7 @@ apply_mrtr_params!(params::Dict{String,Any}, input_responses, request_state) = b end function custom_tool_headers(client::MCPClient, name::String, arguments) + client.transport.kind == :stdio && return HeaderPair[] client_is_modern(client) || return HeaderPair[] if !haskey(client.tool_schemas, name) cursor = nothing @@ -202,6 +203,7 @@ function listen_subscriptions!( resource_uris=String[], headers=nothing, ) + ensure_http_transport(client.transport) client_is_modern(client) || throw(mcp_error(:unsupported_protocol_version, "subscriptions/listen requires protocol version >= $(PROTOCOL_VERSION_2026_07_28)")) notifications = Dict{String,Any}() tools_list_changed && (notifications["toolsListChanged"] = true) @@ -303,11 +305,16 @@ function initialize_client!( client::MCPClient; protocol_version::AbstractString=client.protocol_version, capabilities=nothing, - client_info=default_client_info(), + client_info=client.transport.kind == :stdio ? client.client_info : default_client_info(), extra_params=nothing, headers=nothing, timeout_ms=nothing, ) + if client.transport.kind == :stdio + String(protocol_version) == client.protocol_version || + throw(ArgumentError("Choose the stdio protocol version in MCPClientConfig before starting the child")) + capabilities === nothing && (capabilities = client.capabilities) + end if client_is_modern(client) # Modern protocol has no initialize handshake; record identity for # per-request _meta and use server/discover for capability discovery. @@ -327,6 +334,11 @@ function initialize_client!( merge_extra_params!(params, extra_params) result = jsonrpc_call(client, JSONRPC_METHOD_INITIALIZE; params=params, headers=headers, timeout_ms=timeout_ms) session_data = result isa AbstractDict ? to_json_dict(result) : Dict{String,Any}() + if client.transport.kind == :stdio + get(session_data, "protocolVersion", nothing) == String(protocol_version) || + throw(mcp_error(:unsupported_protocol_version, "The stdio server did not accept protocol version $(protocol_version)")) + client.protocol_version = String(protocol_version) + end client.session = session_data client.last_event_id = nothing send_initialized_notification!(client; headers=headers) @@ -341,6 +353,7 @@ function cancel_request(client::MCPClient, request_id; reason=nothing, headers=n end function open_event_stream(client::MCPClient; headers=nothing, timeout=nothing) + ensure_http_transport(client.transport) client_is_modern(client) && throw(mcp_error(:unsupported_protocol_version, "The 2026-07-28 protocol uses subscriptions/listen instead of a standalone event stream")) client.initialized || throw(mcp_error(:not_initialized, "Client must be initialized before opening an event stream")) header_pairs = normalize_headers(headers) @@ -459,6 +472,10 @@ function send_jsonrpc_response!(client::MCPClient, id; result=nothing, error=not payload["error"] = normalize_jsonrpc_response_error(error) end body = JSON.json(payload) + if client.transport.kind == :stdio + stdio_check_headers(client, headers) + return stdio_write!(client, body, stdio_deadline(client, timeout, nothing)) + end response = submit_jsonrpc_request(client, body; headers=normalize_headers(headers), timeout=timeout) return response end @@ -584,6 +601,7 @@ function event_listener_loop(client::MCPClient, poll_interval::Real, headers) end function start_event_listener!(client::MCPClient; poll_interval::Real=1.0, headers=nothing) + ensure_http_transport(client.transport) client_is_modern(client) && throw(mcp_error(:unsupported_protocol_version, "The 2026-07-28 protocol uses listen_subscriptions! instead of the legacy event listener")) client.initialized || throw(mcp_error(:not_initialized, "Client must be initialized before starting event listener")) stop_event_listener!(client) @@ -615,6 +633,11 @@ function stop_event_listener!(client::MCPClient) end function terminate_session!(client::MCPClient; headers=nothing, timeout=nothing) + if client.transport.kind == :stdio + stdio_check_headers(client, headers) + seconds = timeout === nothing ? 5.0 : stdio_timeout_seconds(client, timeout, nothing) + return close(client; timeout=seconds) + end if client_is_modern(client) stop_event_listener!(client) client.session = nothing diff --git a/src/jsonrpc.jl b/src/jsonrpc.jl index 48cca41..3b3d4ff 100644 --- a/src/jsonrpc.jl +++ b/src/jsonrpc.jl @@ -72,6 +72,9 @@ function jsonrpc_call( timeout=nothing, timeout_ms=nothing, ) + if client.transport.kind == :stdio + return stdio_jsonrpc_call(client, method; params, notification, headers, timeout, timeout_ms) + end ensure_http_transport(client.transport) ensure_client_readiness(client, String(method), notification) method_str = String(method) @@ -273,7 +276,7 @@ end function ensure_client_readiness(client::MCPClient, method::AbstractString, notification::Bool) if client_is_modern(client) - method == JSONRPC_METHOD_NOTIFICATIONS_CANCELLED && + method == JSONRPC_METHOD_NOTIFICATIONS_CANCELLED && client.transport.kind == :http && throw(mcp_error(:unsupported_protocol_version, "Streamable HTTP cancellation in the 2026-07-28 protocol closes the response stream")) method in LEGACY_ONLY_METHODS && throw(mcp_error(:unsupported_protocol_version, "$(method) is not part of the 2026-07-28 protocol")) return @@ -281,7 +284,8 @@ function ensure_client_readiness(client::MCPClient, method::AbstractString, noti if method == JSONRPC_METHOD_INITIALIZE return elseif method == JSONRPC_METHOD_NOTIFICATIONS_INITIALIZED - client.session_id === nothing && throw(mcp_error(:session_required, "Cannot send notifications/initialized before establishing a session")) + ready = client.transport.kind == :stdio ? client.session !== nothing : client.session_id !== nothing + ready || throw(mcp_error(:session_required, "Cannot send notifications/initialized before establishing a session")) return end client.initialized || throw(mcp_error(:not_initialized, "Client must complete initialization before calling $(method)")) diff --git a/src/stdio.jl b/src/stdio.jl new file mode 100644 index 0000000..779d60b --- /dev/null +++ b/src/stdio.jl @@ -0,0 +1,453 @@ +""" + ModelContextProtocol.prepare_stdio_client(command::Cmd; config=MCPClientConfig(), + stderr=Base.stderr, capabilities=Dict(), client_info=default_client_info(), + max_message_bytes=16*1024*1024, max_pending_messages=128) + +Launch and own a local MCP server process. Use `initialize_client!` and the +ordinary client call APIs, then `close(client)` or `terminate_session!`. The +do-block form closes the client even when its body throws. Register handlers +and finish initialization before making concurrent calls. + +Protocol selection is explicit: `2025-11-25` uses initialization and permits +server requests; `2026-07-28` uses discovery and per-request metadata. HTTP +headers, OAuth, HTTP event streams, and modern subscriptions are unsupported. +There is no automatic protocol fallback, process restart, or request replay. +Custom HTTP adapters and verbose HTTP logging are rejected. + +`config.timeout.readtimeout` (seconds, positive and finite) bounds transport +waits, including queued writes; per-call `timeout_ms` overrides it. The HTTP +`connecttimeout` setting has no effect on process startup. Other timeout keys +are rejected. A response timeout sends cancellation; a write timeout breaks +the connection because a partial frame cannot be retried safely. + +The byte limit applies to each incoming and outgoing message. The pending +limit bounds calls, queued writes, and queued callbacks separately. Callback +overflow or invalid server output breaks the connection. Callbacks run in +order on a separate task, so a slow or reentrant handler does not stop response +routing. Closing discards callbacks that have not started. A blocked user +callback must cooperate with shutdown; see `close(::MCPClient)`. + +The child's stderr is redirected directly to a filename, `IOStream`, terminal, +pipe, or `devnull`, separately from protocol stdout. In-memory and custom IO +destinations are unsupported. Caller-provided IO remains caller-owned. +""" +function prepare_stdio_client( + command::Cmd; + config::MCPClientConfig=MCPClientConfig(), + stderr::Union{IO,AbstractString}=Base.stderr, + capabilities=Dict{String,Any}(), + client_info=default_client_info(), + max_message_bytes::Integer=16 * 1024 * 1024, + max_pending_messages::Integer=128, +) + config.transport in (nothing, :stdio) || throw(ArgumentError("stdio requires transport=nothing or :stdio")) + config.http === HTTP || throw(ArgumentError("stdio does not use a custom HTTP adapter")) + config.verbose && throw(ArgumentError("stdio does not support verbose HTTP logging")) + config.protocol_version in (DEFAULT_PROTOCOL_VERSION, PROTOCOL_VERSION_2026_07_28) || + throw(ArgumentError("stdio supports protocol versions $(DEFAULT_PROTOCOL_VERSION) and $(PROTOCOL_VERSION_2026_07_28)")) + max_message_bytes > 0 || throw(ArgumentError("max_message_bytes must be positive")) + max_pending_messages > 0 || throw(ArgumentError("max_pending_messages must be positive")) + while stderr isa IOContext + stderr = stderr.io + end + stderr isa Union{AbstractString,IOStream,Base.TTY,Base.Pipe,Base.PipeEndpoint,Base.DevNull} || + throw(ArgumentError("stdio stderr requires a filename, OS stream, pipe, or devnull")) + byte_limit, pending_limit = Int(max_message_bytes), Int(max_pending_messages) + descriptor = MCPTransportDescriptor(kind=:stdio, url="stdio") + discovery = MCPDiscovery(manifest=JSONDict(), transports=[descriptor], default_transport=descriptor) + client = prepare_manual_client(discovery; config, transport=descriptor, capabilities, client_info) + stdio_check_headers(client, nothing) + stdio_timeout_seconds(client, nothing, nothing) + process = open(pipeline(command; stderr), "r+") + guard = ReentrantLock() + connection = StdioConnection(process, guard, Threads.Condition(guard), Dict{String,Channel{Any}}(), + Channel{StdioWrite}(pending_limit), Channel{JSONDict}(pending_limit), 0, 0, + nothing, nothing, nothing, nothing, nothing, nothing, byte_limit, pending_limit) + client.stdio = connection + connection.writer = @async stdio_writer(client) + connection.callbacks = @async stdio_callbacks(client) + connection.reader = @async stdio_reader(client) + return client +end + +function prepare_stdio_client(f::Function, command::Cmd; kwargs...) + client = prepare_stdio_client(command; kwargs...) + try + return f(client) + finally + close(client) + end +end + +function stdio_connection(client::MCPClient) + connection = client.stdio + connection === nothing && throw(mcp_error(:transport_closed, "Use prepare_stdio_client to own a stdio process")) + return connection +end + +function stdio_check_headers(client::MCPClient, headers) + isempty(normalize_headers(headers)) && isempty(client.headers) && client.auth_token === nothing || + throw(ArgumentError("stdio has no HTTP headers or OAuth bearer token; configure the child command instead")) +end + +function stdio_timeout_seconds(client::MCPClient, timeout, timeout_ms) + settings = normalize_timeout(client, timeout) + all(key -> key in (:connecttimeout, :readtimeout), keys(settings)) || + throw(ArgumentError("stdio timeout supports readtimeout and ignores connecttimeout")) + seconds = timeout_ms === nothing ? get(settings, :readtimeout, JSONRPC_TIMEOUT.readtimeout) : + normalize_timeout_ms(timeout_ms) / 1000 + seconds isa Real && isfinite(seconds) && seconds > 0 || + throw(ArgumentError("stdio readtimeout must be positive and finite")) + return Float64(seconds) +end + +stdio_now() = time_ns() / 1.0e9 +stdio_deadline(client, timeout, timeout_ms) = stdio_now() + stdio_timeout_seconds(client, timeout, timeout_ms) + +function stdio_wait(connection::StdioConnection, channel::Channel, deadline::Float64) + timer = nothing + lock(connection.lock) + try + isready(channel) && return take!(channel) + connection.failure === nothing || throw(connection.failure) + remaining = deadline - stdio_now() + remaining > 0 || throw(mcp_error(:request_timeout, "The stdio request deadline expired")) + expired = Ref(false) + timer = Timer(remaining) do _ + @lock connection.lock begin + expired[] = true + notify(connection.changed; all=true) + end + end + while !isready(channel) && connection.failure === nothing && !expired[] + wait(connection.changed) + end + isready(channel) && return take!(channel) + connection.failure === nothing || throw(connection.failure) + throw(mcp_error(:request_timeout, "The stdio request deadline expired")) + finally + unlock(connection.lock) + timer === nothing || close(timer) + end +end + +function stdio_enqueue!(client::MCPClient, body::String, deadline::Float64) + connection = stdio_connection(client) + ncodeunits(body) <= connection.max_message_bytes || throw(mcp_error(:message_too_large, "Outgoing stdio message exceeds max_message_bytes")) + occursin('\n', body) && throw(mcp_error(:jsonrpc_error, "A stdio message cannot contain a literal newline")) + job = StdioWrite(body, Channel{Any}(1), deadline) + @lock connection.lock begin + connection.failure === nothing || throw(connection.failure) + connection.queued_writes < connection.max_pending_messages || + throw(mcp_error(:transport_busy, "The stdio write queue is full")) + connection.queued_writes += 1 + # This lock reserves a free slot; the writer can only make more room. + put!(connection.outgoing, job) + end + return job +end + +function stdio_write!(client::MCPClient, body::String, deadline::Float64) + connection = stdio_connection(client) + job = stdio_enqueue!(client, body, deadline) + try + result = stdio_wait(connection, job.done, deadline) + result isa Exception && throw(result) + catch err + if err isa MCPError && err.code == :request_timeout + stdio_fail!(connection, err) + end + rethrow() + end + return nothing +end + +function stdio_writer(client::MCPClient) + connection = stdio_connection(client) + try + for job in connection.outgoing + @lock connection.lock begin + connection.failure === nothing || return + connection.queued_writes -= 1 + end + stdio_now() < job.deadline || throw(mcp_error(:request_timeout, "The stdio write deadline expired")) + completed = Ref(false) + timer = Timer(max(0.0, job.deadline - stdio_now())) do _ + @lock connection.lock begin + completed[] || stdio_fail!(connection, + mcp_error(:request_timeout, "The stdio write deadline expired")) + end + end + try + write(connection.process.in, job.body, '\n') + flush(connection.process.in) + @lock connection.lock begin + # Completion survives the caller consuming job.done. + completed[] = true + put!(job.done, connection.failure) + notify(connection.changed; all=true) + end + finally + close(timer) + end + end + catch err + stdio_fail!(connection, err) + end +end + +function stdio_jsonrpc_call(client::MCPClient, method::AbstractString; + params, notification, headers, timeout, timeout_ms) + connection = stdio_connection(client) + stdio_check_headers(client, headers) + deadline = stdio_deadline(client, timeout, timeout_ms) + @lock connection.lock connection.failure === nothing || throw(connection.failure) + method = String(method) + method == JSONRPC_METHOD_SUBSCRIPTIONS_LISTEN && + throw(mcp_error(:transport_unsupported, "Modern stdio subscriptions are not implemented")) + ensure_client_readiness(client, method, notification) + payload = JSONDict("jsonrpc" => JSONRPC_VERSION, "method" => method) + normalized = normalize_params(params) + if client_is_modern(client) + normalized === nothing && (normalized = JSONDict()) + normalized isa JSONDict && (normalized = inject_modern_meta!(client, normalized)) + end + normalized === nothing || (payload["params"] = normalized) + if notification + return stdio_write!(client, JSON.json(payload), deadline) + end + reply = Channel{Any}(1) + id = @lock connection.lock begin + connection.failure === nothing || throw(connection.failure) + length(connection.pending) < connection.max_pending_messages || + throw(mcp_error(:transport_busy, "Too many pending stdio requests")) + client.next_id[] += 1 + id = string(client.next_id[]) + connection.pending[id] = reply + id + end + payload["id"] = id + sent = false + try + stdio_write!(client, JSON.json(payload), deadline) + sent = true + data = stdio_wait(connection, reply, deadline) + data isa Exception && throw(data) + return get(validate_jsonrpc_payload(data), "result", nothing) + catch err + if sent && err isa MCPError && err.code == :request_timeout && method != JSONRPC_METHOD_INITIALIZE + # Cancellation has its own bounded write but does not extend the + # expired caller's wait. The writer also watches unattended writes. + cancellation = JSONDict("jsonrpc" => JSONRPC_VERSION, + "method" => JSONRPC_METHOD_NOTIFICATIONS_CANCELLED, + "params" => JSONDict("requestId" => id)) + client_is_modern(client) && inject_modern_meta!(client, cancellation["params"]) + try + stdio_enqueue!(client, JSON.json(cancellation), stdio_deadline(client, nothing, nothing)) + catch cancellation_error + stdio_fail!(connection, cancellation_error) + end + end + rethrow() + finally + @lock connection.lock pop!(connection.pending, id, nothing) + end +end + +function stdio_message!(client::MCPClient, bytes::Vector{UInt8}) + text = String(copy(bytes)) + isvalid(text) || throw(mcp_error(:jsonrpc_error, "Stdio messages must be valid UTF-8")) + raw = try + JSON.parse(text) + catch err + throw(mcp_error(:jsonrpc_error, "Malformed stdio JSON: $(sprint(showerror, err))")) + end + raw isa AbstractDict && get(raw, "jsonrpc", nothing) == JSONRPC_VERSION || + throw(mcp_error(:jsonrpc_error, "Stdio messages must be JSON-RPC 2.0 objects")) + data = to_json_dict(raw) + connection = stdio_connection(client) + if haskey(data, "method") + data["method"] isa AbstractString || throw(mcp_error(:jsonrpc_error, "Invalid stdio method")) + (haskey(data, "result") || haskey(data, "error")) && + throw(mcp_error(:jsonrpc_error, "A stdio request cannot contain a result or error")) + if haskey(data, "id") + client_is_modern(client) && throw(mcp_error(:jsonrpc_error, "Modern MCP servers cannot initiate requests")) + id = data["id"] + (id isa AbstractString || (id isa Integer && !(id isa Bool))) || + throw(mcp_error(:jsonrpc_error, "Invalid server request ID")) + end + @lock connection.lock begin + connection.failure === nothing || return + connection.queued_events < connection.max_pending_messages || + throw(mcp_error(:callback_overflow, "The stdio callback queue is full")) + connection.queued_events += 1 + put!(connection.events, data) + end + else + xor(haskey(data, "result"), haskey(data, "error")) || + throw(mcp_error(:jsonrpc_error, "A stdio response must contain exactly one of result or error")) + id = get(data, "id", nothing) + id isa String || throw(mcp_error(:jsonrpc_error, "Stdio response ID must match the client's string ID")) + @lock connection.lock begin + reply = pop!(connection.pending, id, nothing) + if reply === nothing + # Replies may race cancellation. Only a previously issued, + # identically encoded ID can be ignored as a late response. + number = tryparse(Int, id) + number !== nothing && 0 < number <= client.next_id[] && string(number) == id || + throw(mcp_error(:jsonrpc_error, "Unexpected stdio response ID $(repr(id))")) + else + put!(reply, data) + notify(connection.changed; all=true) + end + end + end + return nothing +end + +function stdio_reader(client::MCPClient) + connection = stdio_connection(client) + input = connection.process.out + chunk = Vector{UInt8}(undef, 8192) + frame = UInt8[] + try + while !eof(input) + count = readbytes!(input, chunk, min(length(chunk), max(1, bytesavailable(input)))) + start = 1 + for i in 1:count + if chunk[i] == 0x0a + length(frame) + i - start <= connection.max_message_bytes || + throw(mcp_error(:message_too_large, "Incoming stdio message exceeds max_message_bytes")) + append!(frame, @view chunk[start:i-1]) + stdio_message!(client, frame) + empty!(frame) + start = i + 1 + end + end + length(frame) + count - start + 1 <= connection.max_message_bytes || + throw(mcp_error(:message_too_large, "Incoming stdio message exceeds max_message_bytes")) + append!(frame, @view chunk[start:count]) + end + isempty(frame) || throw(mcp_error(:jsonrpc_error, "Stdio output ended with an incomplete frame")) + throw(mcp_error(:transport_closed, "The stdio server closed its output")) + catch err + stdio_fail!(connection, err) + end +end + +function stdio_callbacks(client::MCPClient) + connection = stdio_connection(client) + for payload in connection.events + @lock connection.lock begin + connection.failure === nothing || return + connection.queued_events -= 1 + end + try + if get(payload, "method", nothing) == JSONRPC_METHOD_PING && haskey(payload, "id") && + !haskey(client.request_handlers, JSONRPC_METHOD_PING) + send_jsonrpc_response!(client, payload["id"]) + else + Base.invokelatest(handle_jsonrpc_event!, client, payload) + end + catch err + if @lock(connection.lock, connection.failure === nothing) + @warn "Stdio callback failed" exception=(err, catch_backtrace()) + end + end + end +end + +function stdio_fail!(connection::StdioConnection, error::Exception; shutdown_timeout=5.0) + @lock connection.lock begin + if connection.failure === nothing + connection.failure = error + for reply in values(connection.pending) + put!(reply, error) + end + empty!(connection.pending) + close(connection.outgoing) + close(connection.events) + while isready(connection.outgoing) + take!(connection.outgoing) + end + while isready(connection.events) + take!(connection.events) + end + connection.queued_writes = connection.queued_events = 0 + notify(connection.changed; all=true) + end + if connection.shutdown === nothing || + (istaskdone(connection.shutdown) && fetch(connection.shutdown) isa Exception) + connection.shutdown = @async try + stdio_shutdown(connection, shutdown_timeout) + nothing + catch err + err + end + end + end + return nothing +end + +function stdio_shutdown(connection::StdioConnection, timeout::Real) + process = connection.process + # Closing a pipe can wait for a blocked write. It must not prevent the + # separate shutdown task from escalating termination of a non-reading child. + if connection.input_closer === nothing + connection.input_closer = @async close(process.in) + end + interval = timeout / 3 + try + for signal in (nothing, Base.SIGTERM, Base.SIGKILL) + if signal !== nothing && Base.process_running(process) + kill(process, signal) + end + timedwait(() -> Base.process_exited(process), interval; pollint=0.001) === :ok && break + end + Base.process_exited(process) || throw(mcp_error(:shutdown_timeout, "The stdio child did not exit after termination")) + finally + Base.process_exited(process) && wait(process) + close(process.out) + end + return nothing +end + +""" + close(client::MCPClient; timeout=5.0) + +For an owned stdio client, stop accepting calls, close stdin, then terminate +and, if necessary, kill the child. Fail pending calls and wait for owned IO and +callback tasks. `timeout` bounds this wait; unfinished cleanup is retained and +a later `close` may finish it. The caller owns any custom stderr destination. + +Julia cannot safely interrupt arbitrary user callback code. If a callback +remains blocked after process/IO shutdown, throw `MCPError(:callback_timeout)`; +release the callback and close again. A callback that calls `close` does not +wait on itself, and finishes when its handler returns. Queued callbacks are +discarded. HTTP clients delegate to `terminate_session!`. +""" +function Base.close(client::MCPClient; timeout::Real=5.0) + isfinite(timeout) && timeout > 0 || throw(ArgumentError("close timeout must be positive and finite")) + client.transport.kind == :stdio || return terminate_session!(client; timeout=(readtimeout=timeout,)) + connection = stdio_connection(client) + deadline = stdio_now() + timeout + stdio_fail!(connection, mcp_error(:transport_closed, "The stdio client was closed"); shutdown_timeout=timeout) + task = connection.shutdown + timedwait(() -> istaskdone(task), max(0.0, deadline - stdio_now()); pollint=0.001) === :ok || + throw(mcp_error(:shutdown_timeout, "Stdio process cleanup is still in progress; close again to wait")) + result = fetch(task) + result isa Exception && throw(result) + for worker in (connection.input_closer, connection.reader, connection.writer, connection.callbacks) + worker === nothing && continue + worker === current_task() && continue + if timedwait(() -> istaskdone(worker), max(0.0, deadline - stdio_now()); pollint=0.001) !== :ok + code = worker === connection.callbacks ? :callback_timeout : :shutdown_timeout + throw(mcp_error(code, "A stdio task is still active; release user callback code and close again")) + end + fetch(worker) + end + client.initialized = false + client.session = nothing + client.session_id = nothing + return nothing +end diff --git a/src/types.jl b/src/types.jl index 98db74a..98bcb3c 100644 --- a/src/types.jl +++ b/src/types.jl @@ -258,6 +258,31 @@ Base.@kwdef struct MCPAuthenticationChallenge scopes::Vector{String} end +struct StdioWrite + body::String + done::Channel{Any} + deadline::Float64 +end + +mutable struct StdioConnection + process::Base.Process + lock::ReentrantLock + changed::Threads.Condition + pending::Dict{String,Channel{Any}} + outgoing::Channel{StdioWrite} + events::Channel{JSONDict} + queued_writes::Int + queued_events::Int + failure::Union{Nothing,Exception} + reader::Union{Nothing,Task} + writer::Union{Nothing,Task} + callbacks::Union{Nothing,Task} + input_closer::Union{Nothing,Task} + shutdown::Union{Nothing,Task} + max_message_bytes::Int + max_pending_messages::Int +end + mutable struct MCPClient manifest::JSONDict transport::MCPTransportDescriptor @@ -278,16 +303,20 @@ mutable struct MCPClient event_task::Union{Task,Nothing} last_event_id::Union{String,Nothing} tool_schemas::Dict{String,Dict{String,Any}} + stdio::Union{Nothing,StdioConnection} end # Preserve the public positional client constructors that predate modern -# request metadata and the internal tool-schema cache. +# request metadata, the internal tool-schema cache, and stdio ownership. MCPClient(manifest, transport, protocol_version, http, headers, timeout, verbose, auth_token, session, session_id, initialized, next_id, notification_handlers, request_handlers, event_task, last_event_id) = MCPClient(manifest, transport, protocol_version, http, headers, timeout, verbose, Dict{String,Any}(), Dict{String,Any}(), auth_token, session, session_id, initialized, next_id, notification_handlers, request_handlers, event_task, last_event_id, Dict{String,Dict{String,Any}}()) MCPClient(manifest, transport, protocol_version, http, headers, timeout, verbose, capabilities, client_info, auth_token, session, session_id, initialized, next_id, notification_handlers, request_handlers, event_task, last_event_id) = MCPClient(manifest, transport, protocol_version, http, headers, timeout, verbose, capabilities, client_info, auth_token, session, session_id, initialized, next_id, notification_handlers, request_handlers, event_task, last_event_id, Dict{String,Dict{String,Any}}()) +MCPClient(manifest, transport, protocol_version, http, headers, timeout, verbose, capabilities, client_info, auth_token, session, session_id, initialized, next_id, notification_handlers, request_handlers, event_task, last_event_id, tool_schemas) = + MCPClient(manifest, transport, protocol_version, http, headers, timeout, verbose, capabilities, client_info, auth_token, session, session_id, initialized, next_id, notification_handlers, request_handlers, event_task, last_event_id, tool_schemas, nothing) + # A live subscriptions/listen stream (2026-07-28): notifications matching the # opted-in filter are pushed onto the channel by the server broadcast helpers. struct MCPSubscriptionListener diff --git a/test/runtests.jl b/test/runtests.jl index 19846e7..58dc686 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -1757,4 +1757,5 @@ end end end end +include("stdio.jl") include("trim_compile_tests.jl") diff --git a/test/stdio.jl b/test/stdio.jl new file mode 100644 index 0000000..e08e363 --- /dev/null +++ b/test/stdio.jl @@ -0,0 +1,486 @@ +const StdioMCP = ModelContextProtocol +using Logging + +struct StdioBlockingLogger <: AbstractLogger + started::Channel{Nothing} + release::Channel{Nothing} +end +Logging.min_enabled_level(::StdioBlockingLogger) = Logging.Warn +Logging.shouldlog(::StdioBlockingLogger, args...) = true +Logging.catch_exceptions(::StdioBlockingLogger) = false +function Logging.handle_message(logger::StdioBlockingLogger, args...; kwargs...) + put!(logger.started, nothing) + take!(logger.release) +end + +struct StdioGatedFlush <: IO + input::IO + flushing::Channel{Nothing} + release::Base.Event +end +Base.isopen(io::StdioGatedFlush) = isopen(io.input) +Base.close(io::StdioGatedFlush) = close(io.input) +Base.unsafe_write(io::StdioGatedFlush, pointer::Ptr{UInt8}, count::UInt) = unsafe_write(io.input, pointer, count) +Base.write(io::StdioGatedFlush, byte::UInt8) = write(io.input, byte) +function Base.flush(io::StdioGatedFlush) + put!(io.flushing, nothing) + wait(io.release) + return nothing +end + +function stdio_test_command(mode="normal") + if mode == "ignore_signals" && !Sys.iswindows() + # An OS process with inherited SIG_IGN avoids Julia's own signal + # handling. exec retains the owned PID and leaves no helper child. + script = raw""" + trap '' TERM + IFS= read -r request + printf '%s\n' '{"jsonrpc":"2.0","id":"1","result":{"protocolVersion":"2025-11-25","capabilities":{},"serverInfo":{"name":"signal-test","version":"1"}}}' + IFS= read -r initialized + printf '%s\n' '{"jsonrpc":"2.0","method":"test/ready","params":{}}' + exec sleep 60 + """ + return `sh -c $script` + end + return `$(Base.julia_cmd()) --startup-file=no --threads=1 --project=$(dirname(Base.active_project())) $(joinpath(@__DIR__, "stdio_peer.jl")) $mode` +end + +function stdio_test_client(mode="normal"; protocol="2025-11-25", kwargs...) + return StdioMCP.prepare_stdio_client(stdio_test_command(mode); + config=MCPClientConfig(protocol_version=protocol, timeout=(readtimeout=20,)), kwargs...) +end + +function stdio_take(channel) + timedwait(() -> isready(channel), 20; pollint=0.001) === :ok || error("stdio test signal timed out") + return take!(channel) +end + +stdio_catch(f) = try + f() +catch err + err +end + +function stdio_test_closed(client) + close(client; timeout=10) + connection = client.stdio + @test Base.process_exited(connection.process) + @test !isopen(connection.process.in) && !isopen(connection.process.out) + @test isempty(connection.pending) + @test connection.queued_events == connection.queued_writes == 0 + @test all(task -> task === nothing || istaskdone(task), + (connection.reader, connection.writer, connection.callbacks, connection.input_closer, connection.shutdown)) + @test close(client) === nothing +end + +@testset "Owned stdio client" begin + @testset "Protocol $protocol" for protocol in ("2025-11-25", "2026-07-28") + notifications = Channel{Any}(4) + cancelled = Channel{Any}(2) + client = stdio_test_client(; protocol, + capabilities=Dict("roots"=>Dict()), client_info=Dict("name"=>"owned-test", "version"=>"1")) + register_notification_handler!(client, "test/event", (_, _, value) -> put!(notifications, value)) + register_notification_handler!(client, "test/cancelled", (_, _, value) -> put!(cancelled, value)) + register_request_handler!(client, "test/request", (c, _, value, _) -> + call_tool(c, "echo"; arguments=value)["structuredContent"]) + register_request_handler!(client, "test/error", (_, _, _, _) -> error("handler failed")) + try + initialized = initialize_client!(client) + if protocol == "2025-11-25" + @test initialized["received"]["clientInfo"]["name"] == "owned-test" + @test haskey(initialized["received"]["capabilities"], "roots") + @test client.session_id === nothing + else + @test initialized["supportedVersions"] == [protocol] + end + @test list_tools(client)["tools"][1]["name"] == "echo" + args = Dict("text"=>"line one\nλ\0line two", "array"=>Any[1, false, nothing]) + result = call_tool(client, "echo"; arguments=args) + @test result["structuredContent"] == args + protocol == "2026-07-28" && @test result["received"]["_meta"][StdioMCP.META_PROTOCOL_VERSION] == protocol + @test get_prompt(client, "example")["received"]["method"] == "prompts/get" + @test read_resource(client, "test://resource")["received"]["method"] == "resources/read" + interim = call_tool(client, "input_required") + @test StdioMCP.is_input_required(interim) + @test interim["requestState"] == "state" + call_tool(client, "notify"; arguments=Dict("value"=>1)) + @test stdio_take(notifications)["value"] == 1 + error = stdio_catch(() -> call_tool(client, "error")) + @test error isa MCPError + @test occursin("-32602", sprint(showerror, error)) + @test occursin("field", sprint(showerror, error)) + + reversed = [@async(call_tool(client, "reverse"; arguments=Dict("value"=>i))) for i in 1:2] + @test [fetch(task)["value"] for task in reversed] == [1, 2] + concurrent = [Threads.@spawn(call_tool(client, "echo"; arguments=Dict("value"=>i))) for i in 1:16] + @test [fetch(task)["structuredContent"]["value"] for task in concurrent] == collect(1:16) + + if protocol == "2025-11-25" + # An integer server ID and the client's string IDs occupy + # separate directions. The handler calls back into the client. + callback = call_tool(client, "server_request"; arguments=Dict("id"=>1, "value"=>42)) + @test callback["callback"]["id"] === 1 + @test callback["callback"]["result"]["value"] == 42 + missing = call_tool(client, "server_request"; arguments=Dict("id"=>"unknown", "method"=>"missing")) + @test missing["callback"]["error"]["code"] == -32601 + failed = call_tool(client, "server_request"; arguments=Dict("method"=>"test/error")) + @test failed["callback"]["error"]["code"] == -32603 + @test occursin("handler failed", failed["callback"]["error"]["message"]) + @test haskey(call_tool(client, "server_request"; arguments=Dict("method"=>"ping"))["callback"], "result") + end + + timed = stdio_catch(() -> call_tool(client, "hang"; timeout_ms=50)) + @test timed isa MCPError && timed.code == :request_timeout + @test stdio_take(cancelled)["requestId"] isa String + @test call_tool(client, "echo"; arguments=Dict("after"=>"late response"))["structuredContent"]["after"] == "late response" + @test isempty(client.stdio.pending) + @test_throws ArgumentError call_tool(client, "echo"; headers=["X-Test"=>"no"]) + @test_throws ArgumentError call_tool(client, "echo"; timeout_ms=0) + @test_throws ArgumentError StdioMCP.jsonrpc_call(client, "tools/list"; timeout=(retry=true,)) + @test_throws MCPError open_event_stream(client) + @test_throws MCPError start_event_listener!(client) + @test_throws MCPError StdioMCP.listen_subscriptions!(client) + + # Retain each existing positional construction arity. + fields = ntuple(i -> getfield(client, i), 19) + @test MCPClient(fields...).stdio === nothing + @test MCPClient(fields[1:18]...).stdio === nothing + @test MCPClient(fields[1:7]..., fields[10:18]...).stdio === nothing + finally + stdio_test_closed(client) + end + end + + @testset "Configuration and ownership" begin + absent = Cmd(["not-an-installed-mcp-test-command"]) + for config in (MCPClientConfig(transport=:http), MCPClientConfig(protocol_version="unknown"), + MCPClientConfig(http=Base), MCPClientConfig(verbose=true), + MCPClientConfig(headers=["X-Test"=>"no"]), MCPClientConfig(timeout=(readtimeout=0,)), + MCPClientConfig(timeout=(readtimeout=Inf,)), MCPClientConfig(timeout=(retry=true,))) + @test_throws ArgumentError StdioMCP.prepare_stdio_client(absent; config) + end + @test_throws ArgumentError StdioMCP.prepare_stdio_client(absent; max_message_bytes=0) + @test_throws ArgumentError StdioMCP.prepare_stdio_client(absent; max_pending_messages=0) + @test_throws ArgumentError StdioMCP.prepare_stdio_client(absent; stderr=IOBuffer()) + @test_throws Base.IOError StdioMCP.prepare_stdio_client(absent) + client = stdio_test_client("wrong_version"; stderr=devnull) + try + @test_throws ArgumentError initialize_client!(client; protocol_version="2026-07-28") + error = stdio_catch(() -> initialize_client!(client)) + @test error isa MCPError && error.code == :unsupported_protocol_version + @test !client.initialized && client.session === nothing + finally + stdio_test_closed(client) + end + owned = Ref{Any}() + original = ErrorException("do-block body failed") + error = stdio_catch() do + StdioMCP.prepare_stdio_client(stdio_test_command(); stderr=devnull) do client + owned[] = client + initialize_client!(client) + throw(original) + end + end + @test error === original + stdio_test_closed(owned[]) + client = stdio_test_client(; max_message_bytes=2048, stderr=devnull) + try + initialize_client!(client) + error = stdio_catch(() -> call_tool(client, "echo"; arguments=Dict("large"=>repeat("x", 4096)))) + @test error isa MCPError && error.code == :message_too_large + @test isempty(client.stdio.pending) + @test call_tool(client, "echo"; arguments=Dict("still"=>"usable"))["structuredContent"]["still"] == "usable" + finally + stdio_test_closed(client) + end + end + + @testset "Protocol failure $name" for (name, code) in + (("bad_id", :jsonrpc_error), ("unknown_id", :jsonrpc_error), + ("malformed", :jsonrpc_error), ("invalid_utf8", :jsonrpc_error), + ("partial", :jsonrpc_error), ("eof", :transport_closed), ("oversized", :message_too_large)) + client = stdio_test_client(; max_message_bytes=2048, stderr=devnull) + try + initialize_client!(client) + error = stdio_catch(() -> call_tool(client, name)) + @test error isa MCPError && error.code == code + @test_throws MCPError call_tool(client, "echo") + finally + stdio_test_closed(client) + end + name == "eof" && @test client.stdio.process.exitcode == 7 + end + + @testset "Modern server calls are rejected" begin + client = stdio_test_client(; protocol="2026-07-28", stderr=devnull) + try + initialize_client!(client) + @test_throws MCPError call_tool(client, "server_request") + finally + stdio_test_closed(client) + end + end + + @testset "A completed response remains valid before EOF" begin + client = stdio_test_client(; stderr=devnull) + try + initialize_client!(client) + @test call_tool(client, "reply_then_eof")["complete"] + finally + stdio_test_closed(client) + end + end + + @testset "Stderr is separate and borrowed" begin + mktemp() do _, output + client = stdio_test_client(; stderr=output) + try + initialize_client!(client) + @test call_tool(client, "stderr")["ok"] + finally + stdio_test_closed(client) + end + @test isopen(output) + seekstart(output) + @test length(readlines(output)) == 10000 + end + end + + @testset "A blocked callback cannot block response routing" begin + started, release = Channel{Nothing}(1), Channel{Nothing}(1) + client = stdio_test_client(; max_pending_messages=2, stderr=devnull) + register_notification_handler!(client, "test/event", (_, _, _) -> (put!(started, nothing); take!(release))) + try + initialize_client!(client) + call_tool(client, "notify") + stdio_take(started) + @test call_tool(client, "echo"; arguments=Dict("routed"=>true))["structuredContent"]["routed"] + error = stdio_catch(() -> call_tool(client, "flood")) + @test error isa MCPError && error.code == :callback_overflow + closed = stdio_catch(() -> close(client; timeout=0.5)) + @test closed isa MCPError && closed.code == :callback_timeout + @test Base.process_exited(client.stdio.process) + @test !istaskdone(client.stdio.callbacks) + finally + put!(release, nothing) + stdio_test_closed(client) + end + end + + @testset "A blocked stderr sink does not prevent shutdown" begin + output = Pipe() + Base.link_pipe!(output; reader_supports_async=true, writer_supports_async=true) + client = stdio_test_client(; stderr=output) + try + initialize_client!(client) + error = stdio_catch(() -> call_tool(client, "stderr"; timeout_ms=300)) + @test error isa MCPError && error.code == :request_timeout + stdio_test_closed(client) + @test isopen(output.in) && isopen(output.out) + finally + close(output) + stdio_test_closed(client) + end + end + + @testset "Callback logging does not hold the routing lock" begin + logger = StdioBlockingLogger(Channel{Nothing}(1), Channel{Nothing}(1)) + client = with_logger(logger) do + stdio_test_client(; max_message_bytes=2048, stderr=devnull) + end + register_request_handler!(client, "test/request", (_, _, _, _) -> Base.error(repeat("x", 4096))) + outer = nothing + try + initialize_client!(client) + outer = @async stdio_catch(() -> call_tool(client, "server_request")) + stdio_take(logger.started) + routed = @async call_tool(client, "echo"; arguments=Dict("routed"=>true)) + finished = timedwait(() -> istaskdone(routed), 3; pollint=0.001) === :ok + @test finished + if finished + @test fetch(routed)["structuredContent"]["routed"] + error = stdio_catch(() -> close(client; timeout=0.5)) + @test error isa MCPError && error.code == :callback_timeout + end + finally + put!(logger.release, nothing) + stdio_test_closed(client) + end + @test fetch(outer) isa MCPError + end + + @testset "Close from a reentrant callback" begin + closed = Channel{Any}(1) + client = stdio_test_client(; stderr=devnull) + register_request_handler!(client, "test/request", (c, _, _, _) -> begin + nested = call_tool(c, "echo"; arguments=Dict("nested"=>true))["structuredContent"]["nested"] + close(c) + put!(closed, nested) + nothing + end) + try + initialize_client!(client) + @test_throws MCPError call_tool(client, "server_request") + @test stdio_take(closed) + finally + stdio_test_closed(client) + end + end + + @testset "Reentrant calls race $action" for action in ("eof", "close") + nested = Channel{Any}(1) + client = stdio_test_client(; stderr=devnull) + register_request_handler!(client, "test/request", (c, _, _, _) -> begin + put!(nested, stdio_catch(() -> call_tool(c, "hang"))) + Dict() + end) + try + initialize_client!(client) + outer = @async stdio_catch(() -> call_tool(client, "server_request")) + @test timedwait(() -> lock(() -> length(client.stdio.pending) == 2, client.stdio.lock), + 20; pollint=0.001) === :ok + if action == "eof" + @test_throws MCPError call_tool(client, "eof") + else + close(client) + end + @test fetch(outer) isa MCPError + @test stdio_take(nested) isa MCPError + finally + stdio_test_closed(client) + end + end + + @testset "Bounded pending calls and close" begin + waiting = Channel{Any}(2) + client = stdio_test_client(; max_pending_messages=2, stderr=devnull) + register_notification_handler!(client, "test/waiting", (_, _, params) -> put!(waiting, params)) + initialize_client!(client) + calls = [@async(stdio_catch(() -> call_tool(client, "hang"))) for _ in 1:2] + stdio_take(waiting) + stdio_take(waiting) + error = stdio_catch(() -> call_tool(client, "echo")) + @test error isa MCPError && error.code == :transport_busy + stdio_test_closed(client) + @test timedwait(() -> all(istaskdone, calls), 1; pollint=0.001) === :ok + @test all(task -> fetch(task) isa MCPError, calls) + end + + @testset "One deadline does not expire another call" begin + waiting = Channel{Any}(2) + client = stdio_test_client(; stderr=devnull) + register_notification_handler!(client, "test/waiting", (_, _, value) -> put!(waiting, value)) + try + initialize_client!(client) + short = @async stdio_catch(() -> call_tool(client, "hang"; + arguments=Dict("tag"=>"short"), timeout_ms=100)) + long = @async stdio_catch(() -> call_tool(client, "hang"; + arguments=Dict("tag"=>"long"), timeout_ms=5000)) + messages = [stdio_take(waiting), stdio_take(waiting)] + long_id = only(message["id"] for message in messages if message["tag"] == "long") + expired = fetch(short) + @test expired isa MCPError && expired.code == :request_timeout + @test !istaskdone(long) + StdioMCP.cancel_request(client, long_id) + @test fetch(long)["late"] + finally + stdio_test_closed(client) + end + end + + @testset "Blocked stdin writes have deadlines" begin + ready = Channel{Nothing}(1) + client = stdio_test_client("blocked_input"; stderr=devnull) + register_notification_handler!(client, "test/ready", (_, _, _) -> put!(ready, nothing)) + try + initialize_client!(client) + stdio_take(ready) + started = time_ns() + error = stdio_catch(() -> call_tool(client, "echo"; + arguments=Dict("large"=>repeat("x", 8 * 1024 * 1024)), timeout_ms=300)) + @test error isa MCPError && error.code == :request_timeout + @test (time_ns() - started) / 1e9 < 3 + finally + stdio_test_closed(client) + end + end + + @testset "A late flush cannot publish success after its deadline" begin + client = stdio_test_client(; stderr=devnull) + gate = nothing + try + initialize_client!(client) + connection = client.stdio + gate = StdioGatedFlush(connection.process.in, Channel{Nothing}(1), Base.Event()) + connection.process.in = gate + body = StdioMCP.JSON.json(Dict("jsonrpc"=>"2.0", "method"=>"notifications/initialized")) + job = StdioMCP.stdio_enqueue!(client, body, StdioMCP.stdio_now() + 1) + stdio_take(gate.flushing) + @test timedwait(() -> lock(() -> connection.failure !== nothing, connection.lock), + 3; pollint=0.001) === :ok + failure = connection.failure + @test failure isa MCPError && failure.code == :request_timeout + # The bytes were written, but flush was held until after the + # actual write timer failed the connection. Publish that failure. + notify(gate.release) + @test timedwait(() -> isready(job.done), 3; pollint=0.001) === :ok + @test StdioMCP.stdio_wait(connection, job.done, StdioMCP.stdio_now() + 1) === failure + finally + gate === nothing || notify(gate.release) + stdio_test_closed(client) + end + end + + @testset "Shutdown escalation $mode" for mode in ("ignore_eof", "ignore_signals") + ready = Channel{Nothing}(1) + client = stdio_test_client(mode; stderr=devnull) + register_notification_handler!(client, "test/ready", (_, _, _) -> put!(ready, nothing)) + initialize_client!(client) + stdio_take(ready) + close(client; timeout=3) + stdio_test_closed(client) + mode == "ignore_signals" && !Sys.iswindows() && @test client.stdio.process.termsignal == Base.SIGKILL + end + + @testset "Completed failed cleanup is retryable" begin + client = stdio_test_client(; stderr=devnull) + initialize_client!(client) + connection = client.stdio + release = Channel{Nothing}(1) + original = StdioMCP.mcp_error(:shutdown_timeout, "Injected failure after child exit") + # Fault injection holds the first close at a real task barrier and + # leaves process cleanup incomplete, as an expired shutdown would. + connection.shutdown = @async begin + take!(release) + connection.input_closer = @async close(connection.process.in) + wait(connection.process) + original + end + first_close = @async stdio_catch(() -> close(client)) + @test timedwait(() -> lock(() -> connection.failure !== nothing, connection.lock), + 20; pollint=0.001) === :ok + put!(release, nothing) + @test fetch(first_close) === original + input_closer = connection.input_closer + stdio_test_closed(client) + @test connection.input_closer === input_closer + end + + @testset "Exit racing escalation" begin + for _ in 1:4 + client = stdio_test_client(; stderr=devnull) + initialize_client!(client) + gate = Channel{Nothing}(2) + exited = @async (take!(gate); stdio_catch(() -> call_tool(client, "eof"))) + closed = @async (take!(gate); stdio_catch(() -> close(client; timeout=0.01))) + put!(gate, nothing) + put!(gate, nothing) + @test fetch(exited) isa MCPError + first_close = fetch(closed) + @test first_close === nothing || (first_close isa MCPError && first_close.code == :shutdown_timeout) + @test timedwait(() -> istaskdone(client.stdio.shutdown) && Base.process_exited(client.stdio.process), + 20; pollint=0.001) === :ok + stdio_test_closed(client) + end + end +end diff --git a/test/stdio_peer.jl b/test/stdio_peer.jl new file mode 100644 index 0000000..0989176 --- /dev/null +++ b/test/stdio_peer.jl @@ -0,0 +1,116 @@ +using ModelContextProtocol: JSON + +const VERSION_KEY = "io.modelcontextprotocol/protocolVersion" +const mode = isempty(ARGS) ? "normal" : only(ARGS) +const output_lock = ReentrantLock() +function emit(message) + lock(output_lock) do + println(stdout, JSON.json(message)) + flush(stdout) + end +end +respond(id, result) = emit(Dict("jsonrpc"=>"2.0", "id"=>id, "result"=>result)) +notify(method, params=Dict()) = emit(Dict("jsonrpc"=>"2.0", "method"=>method, "params"=>params)) +const held = Any[] +const callbacks = Dict{Any,Any}() +initialized = false + +for line in eachline(stdin) + message = JSON.parse(line) + method = get(message, "method", nothing) + id = get(message, "id", nothing) + params = get(message, "params", Dict()) + if method === nothing + original = pop!(callbacks, id) + respond(original, Dict("callback"=>message)) + elseif method == "initialize" + version = mode == "wrong_version" ? "1900-01-01" : params["protocolVersion"] + respond(id, Dict("protocolVersion"=>version, "capabilities"=>Dict("tools"=>Dict()), + "serverInfo"=>Dict("name"=>"local-test-peer", "version"=>"1"), "received"=>params)) + elseif method == "notifications/initialized" + global initialized = true + if mode == "blocked_input" + notify("test/ready") + wait(Condition()) + elseif mode in ("ignore_eof", "ignore_signals") + notify("test/ready") + end + elseif method == "server/discover" + @assert params["_meta"][VERSION_KEY] == "2026-07-28" + respond(id, Dict("supportedVersions"=>["2026-07-28"], "capabilities"=>Dict("tools"=>Dict()), "serverInfo"=>Dict("name"=>"test"))) + elseif method == "notifications/cancelled" + notify("test/cancelled", params) + # A reply already being written can race cancellation. It must not be + # mistaken for the next request, nor close an otherwise usable client. + respond(params["requestId"], Dict("late"=>true)) + elseif method == "tools/list" + @assert initialized || params["_meta"][VERSION_KEY] == "2026-07-28" + respond(id, Dict("tools"=>[Dict("name"=>"echo", "inputSchema"=>Dict("type"=>"object"))], "received"=>params)) + elseif method == "ping" + respond(id, Dict()) + elseif method == "tools/call" + @assert initialized || params["_meta"][VERSION_KEY] == "2026-07-28" + name = params["name"] + arguments = get(params, "arguments", Dict()) + if name == "echo" + respond(id, Dict("structuredContent"=>arguments, "received"=>params)) + elseif name == "reverse" + push!(held, (id, arguments)) + if length(held) == 2 + for (request_id, value) in reverse(held) + respond(request_id, value) + end + empty!(held) + end + elseif name == "hang" + notify("test/waiting", Dict("id"=>id, "tag"=>get(arguments, "tag", nothing))) + elseif name == "error" + emit(Dict("jsonrpc"=>"2.0", "id"=>id, "error"=>Dict("code"=>-32602, "message"=>"bad argument", "data"=>Dict("field"=>"x")))) + elseif name == "notify" + notify("test/event", arguments) + respond(id, Dict()) + elseif name == "server_request" + request_id = get(arguments, "id", 77) + callbacks[request_id] = id + emit(Dict("jsonrpc"=>"2.0", "id"=>request_id, "method"=>get(arguments, "method", "test/request"), "params"=>arguments)) + elseif name == "flood" + for i in 1:8 + notify("test/event", Dict("value"=>i)) + end + respond(id, Dict()) + elseif name == "stderr" + write(stderr, repeat("diagnostic only\n", 10000)) + flush(stderr) + respond(id, Dict("ok"=>true)) + elseif name == "bad_id" + emit(Dict("jsonrpc"=>"2.0", "id"=>parse(Int, id), "result"=>Dict())) + elseif name == "unknown_id" + respond("never-issued", Dict()) + elseif name == "malformed" + println(stdout, "not json") + flush(stdout) + elseif name == "invalid_utf8" + write(stdout, UInt8[0xff, 0x0a]) + flush(stdout) + elseif name == "oversized" + println(stdout, repeat("x", 4096)) + flush(stdout) + elseif name == "partial" + write(stdout, JSON.json(Dict("jsonrpc"=>"2.0", "id"=>id, "result"=>Dict()))) + flush(stdout) + exit(0) + elseif name == "eof" + exit(7) + elseif name == "reply_then_eof" + respond(id, Dict("complete"=>true)) + exit(0) + elseif name == "input_required" + respond(id, Dict("resultType"=>"input_required", "inputRequests"=>Dict("x"=>Dict("method"=>"sampling/createMessage")), "requestState"=>"state")) + else + error("unrecognized test method $name") + end + else + respond(id, Dict("received"=>message)) + end +end +mode in ("ignore_eof", "ignore_signals") && wait(Condition())