Unix Domain Sockets for MCP transport

[scottslewis]

It would be very helpful to use unix domain sockets (supported in Java 17) for localhost MCP Client <-> MCP Server communication.

Currently, stdio transport is used for testing, debugging, and deployment of MCP servers.

unix domain sockets would provide both better performance (avoid tcp/ip stack), and would free up the use of stdin and stdout in the stdio transport (helpful for testing and debugging).

I can provide the implementation of this if there is interest. I suspect that it will be desirable to create AbstractLocalhostServer/ClientTransport classes, which could then be used as a super class for the current StdioServerTransport and StdioClientTransport...without breaking existing code that uses these...and these abstract super classes could make the implementation of the unix domain socket classes easier as well, since both the stdio client/server transports and the unix socket client/server transport implementations will likely use InputStreams and OutputStreams.

[scottslewis]

I've been working on implementing this and have run into a deadlock on the MCP serverside.

First, I created and bind serverChannel. Then I call

SocketChannel channel = serverChannel.accept();
StdioServerTransportProvider transport = new StdioServerTransportProvider(new ObjectMapper(), new BufferedInputStream(Channels.newInputStream(channel)), new BufferedOutputStream(Channels.newOutputStream(channel)));

The BufferedStreams don't matter.

As you can see, I'm using Channels.newInputStream/newOutputStream(channel) and passing the returned streams to the existing StdioServerTransportProvider constructor.

This all works as expected, until the client sends the initialize message. The server receives the message fine and processes it and finally responds to it and blocks forever in line 268 of StdioServerTransportProvider. Line 268 is the outputStream.flush() in this block

StdioServerTransportProvider$StdioMcpSessionTransport.lambda$startOutboundProcessing$6(McpSchema$JSONRPCMessage, SynchronousSink) line: 268

						 synchronized (outputStream) {
							 outputStream.write(jsonMessage.getBytes(StandardCharsets.UTF_8));
							 outputStream.write("\n".getBytes(StandardCharsets.UTF_8));
							 outputStream.flush();   <-- 268
						 }

As per the stack trace below the channel lock held by the ChannelInputStream attempting to read the next line in

		StdioServerTransportProvider$StdioMcpSessionTransport.lambda$startInboundProcessing$4() line: 209	

Stack trace:

	Thread [pool-3-thread-1] (Suspended)	
		owns: Object  (id=244)	
			waited by: Thread [pool-5-thread-1] (Suspended)	
				owns: Channels$1  (id=249)	
				owns: BufferedOutputStream  (id=238)	
		owns: ChannelInputStream  (id=245)	
		owns: BufferedInputStream  (id=246)	
		owns: InputStreamReader  (id=247)	
		SocketDispatcher.read0(FileDescriptor, long, int) line: not available [native method]	
		SocketDispatcher.read(FileDescriptor, long, int) line: 46	
		IOUtil.readIntoNativeBuffer(FileDescriptor, ByteBuffer, long, boolean, boolean, int, NativeDispatcher) line: 330	
		IOUtil.read(FileDescriptor, ByteBuffer, long, boolean, boolean, int, NativeDispatcher) line: 296	
		IOUtil.read(FileDescriptor, ByteBuffer, long, NativeDispatcher) line: 259	
		SocketChannelImpl.read(ByteBuffer) line: 417	
		ChannelInputStream.read(ReadableByteChannel, ByteBuffer, boolean) line: 59	
		ChannelInputStream.read(ByteBuffer) line: 107	
		ChannelInputStream.read(byte[], int, int) line: 101	
		BufferedInputStream.read1(byte[], int, int) line: 282	
		BufferedInputStream.read(byte[], int, int) line: 343	
		StreamDecoder.readBytes() line: 270	
		StreamDecoder.implRead(char[], int, int) line: 313	
		StreamDecoder.read(char[], int, int) line: 188	
		InputStreamReader.read(char[], int, int) line: 177	
		BufferedReader.fill() line: 162	
		BufferedReader.readLine(boolean, boolean[]) line: 329	
		BufferedReader.readLine() line: 396	
		StdioServerTransportProvider$StdioMcpSessionTransport.lambda$startInboundProcessing$4() line: 209	
		0x0000000800e58c00.run() line: not available	
		SchedulerTask.call() line: 68	
		SchedulerTask.call() line: 28	
		FutureTask<V>.run() line: 264	
		ThreadPoolExecutor.runWorker(ThreadPoolExecutor$Worker) line: 1136	
		ThreadPoolExecutor$Worker.run() line: 635	
		Thread.run() line: 833	
	Thread [pool-5-thread-1] (Suspended)	
		owns: Channels$1  (id=249)	
		owns: BufferedOutputStream  (id=238)	
		waiting for: Object  (id=244)	
			owned by: Thread [pool-3-thread-1] (Suspended)	
		Channels.writeFully(WritableByteChannel, ByteBuffer) line: 90	
		Channels$1.write(byte[], int, int) line: 171	
		BufferedOutputStream.flushBuffer() line: 81	
		BufferedOutputStream.flush() line: 142	
		StdioServerTransportProvider$StdioMcpSessionTransport.lambda$startOutboundProcessing$6(McpSchema$JSONRPCMessage, SynchronousSink) line: 268	
		0x0000000800e64470.accept(Object, Object) line: not available	
		FluxHandleFuseable$HandleFuseableSubscriber<T,R>.tryOnNext(T) line: 119	
		FluxPublishOn$PublishOnConditionalSubscriber<T>.runAsync() line: 1004	
		FluxPublishOn$PublishOnConditionalSubscriber<T>.run() line: 1079	
		WorkerTask.call() line: 84	
		WorkerTask.call() line: 37	
		FutureTask<V>.run() line: 264	
		ThreadPoolExecutor.runWorker(ThreadPoolExecutor$Worker) line: 1136	
		ThreadPoolExecutor$Worker.run() line: 635	
		Thread.run() line: 833	

The blocking InputStream.readLine() has a hold of the channel lock (id=244), the outputStream.flush() cannot get it and so it's deadlocked.

I'm thinking that the way to fix this is create a a new UDSMcpSessionTransport, and pass it the underlying channel (rather than the input streams...and then synchronize on the underlying channel before allowing the having the reader/inputstream or writer/outputstream actually used. I wanted to ask if there might be a better way to handle this.