Add support for Streamable HTTP to dart_mcp

[ghchinoy]

Provide support for the latest transport of the MCP project, Streamable HTTP so that dart/flutter clients can utilize HTTP-based MCP Servers and tools.

[michael-mcroskey]

+1 from FlutterFlow — we're interested in Streamable HTTP support for dart_mcp as we're soon rolling out MCP support

[jakemac53]

Do you need support for authentication as well?

[ksokolovskyi]

Hi @jakemac53, I'm interested in contributing to this issue. Currently, StreamChannel is used for communication between the client and the server. Do you have a specific direction in mind for the Transport API?
Would you be open to a PR addressing this issue?

[jakemac53]

Hi @ksokolovskyi thanks for offering to help here!

My original thought was that we don't need a Transport API at all - I don't know what it provides over just a raw StreamChannel. Instead, I was planning to just provide helper functions for each transport (see lib/stdio.dart for an example), which just give you the stream channel.

But, I am definitely open to any arguments for adding an explicit Transport class if you think we need one.

[jakemac53]

@ksokolovskyi checking if you have any thoughts or have you started looking into this at all?

[ksokolovskyi]

Hi @jakemac53, sorry for the late reply.

I've created a simple MVP to test communication with server via HTTP. It is stateless, so in order to call some tool we have to send all the requests in batch (initialize request, initialized notification, tools list request, concat tool call request).

Example source code

import 'dart:async';
import 'dart:io';
import 'dart:convert';

import 'package:dart_mcp/server.dart';
import 'package:stream_channel/stream_channel.dart';

void main() async {
  final httpServer = await HttpServer.bind(InternetAddress.loopbackIPv4, 3333);
  print('Serving at http://${httpServer.address.host}:${httpServer.port}');

  await for (HttpRequest request in httpServer) {
    // Only process POST requests to the /mcp endpoint.
    if (request.method != 'POST' || request.uri.path != '/mcp') {
      request.response
        ..statusCode = HttpStatus.methodNotAllowed
        ..headers.contentType = ContentType.text
        ..write('Only POST requests to /mcp are allowed.');
      await request.response.close();
      continue;
    }

    // Ensure the Content-Type is application/json.
    final contentType = request.headers.contentType;
    if (contentType == null || contentType.mimeType != 'application/json') {
      request.response
        ..statusCode = HttpStatus.unsupportedMediaType
        ..headers.contentType = ContentType.text
        ..write('Invalid Request: Content-Type must be application/json.');
      await request.response.close();
      continue;
    }

    final responseStreamController = StreamController<String>();

    final channel = StreamChannel<String>(
      Utf8Decoder().bind(request),
      responseStreamController.sink,
    );

    final mcpServer = MCPServerWithTools(channel);

    responseStreamController.stream.listen(
      (response) async {
        request.response
          ..headers.contentType = ContentType.json
          ..statusCode = HttpStatus.ok
          ..write(response);

        await request.response.close();
        print('Responded');
      },
      onDone: () async {
        print('Done');
        await mcpServer.shutdown();
        await responseStreamController.close();
      },
      onError: (error, stackTrace) async {
        print('Error writing RPC response to HTTP: $error\n$stackTrace');

        request.response
          ..statusCode = HttpStatus.internalServerError
          ..headers.contentType = ContentType.json
          ..write(
            jsonEncode({
              'jsonrpc': '2.0',
              'error': {
                'code': -32000,
                'message':
                    'Internal server error while sending response: $error',
              },
            }),
          );
        await request.response.close();
      },
    );
  }
}

/// This server uses the [ToolsSupport] mixin to provide tools to the client.
base class MCPServerWithTools extends MCPServer with ToolsSupport {
  MCPServerWithTools(super.channel, {super.protocolLogSink})
    : super.fromStreamChannel(
        implementation: Implementation(
          name: 'An example dart server with tools support',
          version: '0.1.0',
        ),
        instructions: 'Just list and call the tools :D',
      ) {
    registerTool(concatTool, _concat);
  }

  final concatTool = Tool(
    name: 'concat',
    description: 'concatenates many string parts into one string',
    inputSchema: Schema.object(
      properties: {
        'parts': Schema.list(
          description: 'The parts to concatenate together',
          items: Schema.string(),
        ),
      },
      required: ['parts'],
    ),
  );

  FutureOr<CallToolResult> _concat(CallToolRequest request) => CallToolResult(
    content: [
      TextContent(
        text: (request.arguments!['parts'] as List<dynamic>)
            .cast<String>()
            .join(''),
      ),
    ],
  );
}

Request

curl -X POST http://localhost:3333/mcp \
  -H "Content-Type: application/json" \
  -H "mcp-session-id: test-session-123" \
  -d '[
    {
      "jsonrpc": "2.0",
      "id": 1,
      "method": "initialize",
      "params": {
        "protocolVersion": "2025-03-26",
        "capabilities": {
          "roots": {
            "listChanged": true
          },
          "sampling": {}
        },
        "clientInfo": {
          "name": "ExampleClient",
          "version": "1.0.0"
        }
      }
    },
    {
      "jsonrpc": "2.0",
      "id": 2,
      "method": "notifications/initialized"
    },
    {
      "jsonrpc": "2.0",
      "id": 3,
      "method": "tools/list"
    },
    {
      "jsonrpc": "2.0",
      "id": 4,
      "method": "tools/call",
      "params": {
        "name": "concat",
        "arguments": {
          "parts": [
            "a",
            "b",
            "c"
          ]
        }
      }
    }
  ]'

Response

[
  {
    "jsonrpc": "2.0",
    "result": {
      "protocolVersion": "2025-03-26",
      "capabilities": {
        "tools": {
          "listChanged": true
        }
      },
      "serverInfo": {
        "name": "An example dart server with tools support",
        "version": "0.1.0"
      },
      "instructions": "Just list and call the tools :D"
    },
    "id": 1
  },
  {
    "jsonrpc": "2.0",
    "result": null,
    "id": 2
  },
  {
    "jsonrpc": "2.0",
    "result": {
      "tools": [
        {
          "name": "concat",
          "description": "concatenates many string parts into one string",
          "inputSchema": {
            "type": "object",
            "properties": {
              "parts": {
                "type": "array",
                "description": "The parts to concatenate together",
                "items": {
                  "type": "string"
                }
              }
            },
            "required": [
              "parts"
            ]
          }
        }
      ]
    },
    "id": 3
  },
  {
    "jsonrpc": "2.0",
    "result": {
      "content": [
        {
          "text": "abc",
          "type": "text"
        }
      ]
    },
    "id": 4
  }
]

From what I see now helper functions for HTTP transport would be not enough.

[jakemac53]

From what I see now helper functions for HTTP transport would be not enough.

I am a bit confused, it seems to me that your example demonstrates it would be enough? Ultimately, you were able to create a StreamChannel and give that to the server. So, basically you would just be extracting out the logic to create that stream given an /mcp post request into a helper, something like this:

import 'dart:async';
import 'dart:io';
import 'dart:convert';

import 'package:dart_mcp/server.dart';
import 'package:stream_channel/stream_channel.dart';

StreamChannel<String>? streamableHttpChannel(HttpRequest request) {
  // Only process POST requests to the /mcp endpoint.
  if (request.method != 'POST' || request.uri.path != '/mcp') {
    request.response
      ..statusCode = HttpStatus.methodNotAllowed
      ..headers.contentType = ContentType.text
      ..write('Only POST requests to /mcp are allowed.');
    await request.response.close();
    return null;
  }

  // Ensure the Content-Type is application/json.
  final contentType = request.headers.contentType;
  if (contentType == null || contentType.mimeType != 'application/json') {
    request.response
      ..statusCode = HttpStatus.unsupportedMediaType
      ..headers.contentType = ContentType.text
      ..write('Invalid Request: Content-Type must be application/json.');
    await request.response.close();
    return null;
  }

  final responseStreamController = StreamController<String>();

  final channel = StreamChannel<String>(
    Utf8Decoder().bind(request),
    responseStreamController.sink,
  );

  responseStreamController.stream.listen(
    (response) async {
      request.response
        ..headers.contentType = ContentType.json
        ..statusCode = HttpStatus.ok
        ..write(response);

      await request.response.close();
      print('Responded');
    },
    onDone: () async {
      print('Done');
      await mcpServer.shutdown();
      await responseStreamController.close();
    },
    onError: (error, stackTrace) async {
      print('Error writing RPC response to HTTP: $error\n$stackTrace');

      request.response
        ..statusCode = HttpStatus.internalServerError
        ..headers.contentType = ContentType.json
        ..write(
          jsonEncode({
            'jsonrpc': '2.0',
            'error': {
              'code': -32000,
              'message':
                  'Internal server error while sending response: $error',
            },
          }),
        );
      await request.response.close();
    },
  );

  return channel;
}

void main() async {
  final httpServer = await HttpServer.bind(InternetAddress.loopbackIPv4, 3333);
  print('Serving at http://${httpServer.address.host}:${httpServer.port}');

  await for (HttpRequest request in httpServer) {
    if (streamableHttpChannel(request) case final channel?) {
      MCPServerWithTools(channel);
    }
  }
}

/// This server uses the [ToolsSupport] mixin to provide tools to the client.
base class MCPServerWithTools extends MCPServer with ToolsSupport {
  MCPServerWithTools(super.channel, {super.protocolLogSink})
    : super.fromStreamChannel(
        implementation: Implementation(
          name: 'An example dart server with tools support',
          version: '0.1.0',
        ),
        instructions: 'Just list and call the tools :D',
      ) {
    registerTool(concatTool, _concat);
  }

  final concatTool = Tool(
    name: 'concat',
    description: 'concatenates many string parts into one string',
    inputSchema: Schema.object(
      properties: {
        'parts': Schema.list(
          description: 'The parts to concatenate together',
          items: Schema.string(),
        ),
      },
      required: ['parts'],
    ),
  );

  FutureOr<CallToolResult> _concat(CallToolRequest request) => CallToolResult(
    content: [
      TextContent(
        text: (request.arguments!['parts'] as List<dynamic>)
            .cast<String>()
            .join(''),
      ),
    ],
  );
}

[nikeokoronkwo]

Would love to see this feature.
Would authentication be supported as well for the client and server?

[ksokolovskyi]

I am a bit confused, it seems to me that your example demonstrates it would be enough? Ultimately, you were able to create a StreamChannel and give that to the server. So, basically you would just be extracting out the logic to create that stream given an /mcp post request into a helper, something like this:

Unfortunately, my example works only with batched requests. Below, I share the non-batched requests and the responses that will be received

Non-batched requests

curl -X POST http://localhost:3333/mcp \
  -H "Content-Type: application/json" \
  -H "mcp-session-id: test-session-123" \
  -d '{
      "jsonrpc": "2.0",
      "id": 1,
      "method": "initialize",
      "params": {
        "protocolVersion": "2025-03-26",
        "capabilities": {
          "roots": {
            "listChanged": true
          },
          "sampling": {}
        },
        "clientInfo": {
          "name": "ExampleClient",
          "version": "1.0.0"
        }
      }
    }' && 
curl -X POST http://localhost:3333/mcp \
  -H "Content-Type: application/json" \
  -H "mcp-session-id: test-session-123" \
  -d '{
      "jsonrpc": "2.0",
      "id": 2,
      "method": "notifications/initialized"
    }' && 
curl -X POST http://localhost:3333/mcp \
  -H "Content-Type: application/json" \
  -H "mcp-session-id: test-session-123" \
  -d '{
      "jsonrpc": "2.0",
      "id": 3,
      "method": "tools/list"
    }' && 
curl -X POST http://localhost:3333/mcp \
  -H "Content-Type: application/json" \
  -H "mcp-session-id: test-session-123" \
  -d '{
      "jsonrpc": "2.0",
      "id": 4,
      "method": "tools/call",
      "params": {
        "name": "concat",
        "arguments": {
          "parts": ["a", "b", "c"]
        }
      }
    }'

Non-batched responses

  {
    "jsonrpc": "2.0",
    "result": {
      "protocolVersion": "2025-03-26",
      "capabilities": {
        "tools": {
          "listChanged": true
        }
      },
      "serverInfo": {
        "name": "An example dart server with tools support",
        "version": "0.1.0"
      },
      "instructions": "Just list and call the tools :D"
    },
    "id": 1
  }

  {
    "jsonrpc": "2.0",
    "result": null,
    "id": 2
  }

  {
    "jsonrpc": "2.0",
    "error": {
      "code": -32601,
      "message": "Unknown method \"tools/list\".",
      "data": {
        "request": {
          "jsonrpc": "2.0",
          "id": 3,
          "method": "tools/list"
        }
      }
    },
    "id": 3
  }

  {
    "jsonrpc": "2.0",
    "error": {
      "code": -32601,
      "message": "Unknown method \"tools/call\".",
      "data": {
        "request": {
          "jsonrpc": "2.0",
          "id": 4,
          "method": "tools/call",
          "params": {
            "name": "concat",
            "arguments": {
              "parts": [
                "a",
                "b",
                "c"
              ]
            }
          }
        }
      }
    },
    "id": 4
  }

This happens because we are losing the state of the MCPServerWithTools as we recreate it for each request.

From what I see, to make it work with the current architecture of the package, we need to:

1. Add Mcp-Session-Id support (https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#session-management)
2. Add caching for the server instances for each initialized client according to their Mcp-Session-Id, so we can later reuse the server for all the requests from this client

@jakemac53, what do you think?

[jakemac53]

1. Add Mcp-Session-Id support (https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#session-management)
2. Add caching for the server instances for each initialized client according to their Mcp-Session-Id, so we can later reuse the server for all the requests from this client

Yes, I agree those are the things that we need generally, especially given batch requests are not even a part of the MCP spec any more afaik.

However, if you look at the typescript SDK examples, they don't handle these for you, they just have an example of doing it yourself https://github.com/modelcontextprotocol/typescript-sdk?tab=readme-ov-file#streamable-http.

We could try and handle this though by passing a factory for creating the MCP server instances for each session or something like that, but I think that should probably be an optional thing and we would want the lower level API also exposed.