StreamableHttp - Server transport with state management #553

ihrpr · 2025-04-21T16:20:43Z

Adding support for StreamableHTTP transport transport for MCP servers, providing bidirectional communication over
HTTP with streaming support.

Features

Session management
Implementation for server termination (DELETE)
Includes support for JSON and SSE response modes
Handles Related request IDs for notifications like logging and progress
Example of a server with tool notifications

Follow ups

Suport stateless mode
Support GET SSE
Streamable Http Client

#443

jerome3o-anthropic

Concerns about related_request_id - seems like it's accidentally added. Otherwise looks good

examples/servers/simple-streamablehttp/README.md

jerome3o-anthropic · 2025-04-23T01:34:36Z

src/mcp/server/streamableHttp.py

+        # If no session ID provided but required, return error
+        if not request_session_id:
+            response = self._create_error_response(
+                "Bad Request: Missing session ID",


I get this error when I run the example server:

uv run mcp-simple-streamablehttp --port 3000 --log-level DEBUG

Then

curl http://localhost:3000/mcp/ -d ' { "jsonrpc": "2.0", "method": "initialize", "params": { "protocolVersion": "2025-03-26", "capabilities": {}, "clientInfo": { "name": "curl-client", "version": "1.0.0" } } }' \ -H "Content-Type: application/json" \ -H "Accept: application/json, text/event-stream" \ -X POST | \ jq

output:

{ "jsonrpc": "2.0", "id": "server-error", "error": { "code": -32600, "message": "Bad Request: Missing session ID" } }

But this is before I'm able to get the session from the server to start with

Ah - I was missing the id, thats why. it mistook the message as a notification, and then didn't trigger the if is_initialization_request path. That could have a better error, but idt it should block here

jerome3o-anthropic · 2025-04-23T03:08:52Z

examples/servers/simple-streamablehttp/mcp_simple_streamablehttp/server.py

+
+    app = Server("mcp-streamable-http-demo")
+
+    @app.call_tool()


[nit, also could be a follow up] Would be nice to use fastmcp here - i usually end up going:

server = FastMCP(...) ... server._mcp_server # <- to access the lower level mcp interface

But perhaps we should just make it easier to get to

I think I was quite confused here, we have two folders for servers: fastmcp and servers. I was planning to add fastmcp exaple as a follow up.

fastmcp might also help with related_request_id clarity - it's hidden under log. Low level server does not have a place where we can nicely inject related_request_id

src/mcp/server/streamableHttp.py

src/mcp/shared/session.py

src/mcp/types.py

ihrpr · 2025-04-23T10:07:17Z

@jerome3o-anthropic, very good point on notification params, somehow I completely ignored that fact it's part of the spec 🙈

What do you think about using meta field to propagate related_request_id?

ihrpr · 2025-04-27T20:06:12Z

@jerome3o-anthropic, very good point on notification params, somehow I completely ignored that fact it's part of the spec 🙈

What do you think about using meta field to propagate related_request_id?

I think I'd like to change the approach to use clean separation: #590

jerome3o-anthropic

Approving with the assumption we're gonna tidy up the related_request_id/_meta stuff.

dsp-ant · 2025-04-30T11:12:51Z

examples/servers/simple-streamablehttp/mcp_simple_streamablehttp/server.py

+from mcp.server.streamableHttp import (
+    MCP_SESSION_ID_HEADER,
+    StreamableHTTPServerTransport,
+)


Let's call this mcp.server.http.streamable, or something similar. streamableHttp is not PEP 8 compliant.

dsp-ant · 2025-04-30T11:14:10Z

examples/servers/simple-streamablehttp/mcp_simple_streamablehttp/server.py

+    async def call_tool(
+        name: str, arguments: dict
+    ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
+        ctx = app.request_context
+        interval = arguments.get("interval", 1.0)
+        count = arguments.get("count", 5)
+        caller = arguments.get("caller", "unknown")
+
+        # Send the specified number of notifications with the given interval
+        for i in range(count):
+            await ctx.session.send_log_message(
+                level="info",
+                data=f"Notification {i+1}/{count} from caller: {caller}",
+                logger="notification_stream",
+                # Associates this notification with the original request
+                # Ensures notifications are sent to the correct response stream
+                # Without this, notifications will either go to:
+                # - a standalone SSE stream (if GET request is supported)
+                # - nowhere (if GET request isn't supported)
+                related_request_id=ctx.request_id,
+            )
+            if i < count - 1:  # Don't wait after the last notification
+                await anyio.sleep(interval)
+
+        return [
+            types.TextContent(
+                type="text",
+                text=(
+                    f"Sent {count} notifications with {interval}s interval"
+                    f" for caller: {caller}"
+                ),
+            )
+        ]


We never check the name of the tool that is called. We should.

dsp-ant · 2025-04-30T11:16:30Z

examples/servers/simple-streamablehttp/mcp_simple_streamablehttp/server.py

+    async def handle_streamable_http(scope, receive, send):
+        request = Request(scope, receive)
+        request_mcp_session_id = request.headers.get(MCP_SESSION_ID_HEADER)
+        if (
+            request_mcp_session_id is not None
+            and request_mcp_session_id in server_instances
+        ):
+            transport = server_instances[request_mcp_session_id]
+            logger.debug("Session already exists, handling request directly")
+            await transport.handle_request(scope, receive, send)
+        elif request_mcp_session_id is None:
+            # try to establish new session
+            logger.debug("Creating new transport")
+            # Use lock to prevent race conditions when creating new sessions
+            async with session_creation_lock:
+                new_session_id = uuid4().hex
+                http_transport = StreamableHTTPServerTransport(
+                    mcp_session_id=new_session_id,
+                    is_json_response_enabled=json_response,
+                )
+                server_instances[http_transport.mcp_session_id] = http_transport
+            async with http_transport.connect() as streams:
+                read_stream, write_stream = streams
+
+                async def run_server():
+                    await app.run(
+                        read_stream,
+                        write_stream,
+                        app.create_initialization_options(),
+                    )
+
+                if not task_group:
+                    raise RuntimeError("Task group is not initialized")
+
+                task_group.start_soon(run_server)
+
+                # Handle the HTTP request and return the response
+                await http_transport.handle_request(scope, receive, send)
+        else:
+            response = Response(
+                "Bad Request: No valid session ID provided",
+                status_code=HTTPStatus.BAD_REQUEST,
+            )
+            await response(scope, receive, send)


This feels awfully complex as a lowlevel API, particularly if every MCP Server using the lowlevel API would have to go and use something like this. I wonder if we can provide a more high-level abstraction for this, that makes it more managable.

sure, we can make it happen, will add as a separate PR

dsp-ant · 2025-04-30T11:16:57Z

src/mcp/server/streamableHttp.py

As mentioned above, let's rename this.

@dsp-ant I renamed the files later in the stack, it's quite a few of them by now (like 9 or 10), so was very difficult to rename in the very bottom. Happy to rename in the bottom if that's needed, just let me know.

src/mcp/server/streamableHttp.py

dsp-ant · 2025-04-30T11:27:36Z

src/mcp/server/streamableHttp.py

+        accept_types = [media_type.strip() for media_type in accept_header.split(",")]
+
+        has_json = any(
+            media_type.startswith(CONTENT_TYPE_JSON) for media_type in accept_types
+        )
+        has_sse = any(
+            media_type.startswith(CONTENT_TYPE_SSE) for media_type in accept_types
+        )


Given you have stripped accept_types as a list, you might even put them into a set and just do a check if CONTENT_TYPE_SSE is in media_type or CONTENT_TYPE_JSON is in media_type. It doesn't really matter here, but we don't need to traverse the list 3 times.

dsp-ant · 2025-04-30T11:27:52Z

src/mcp/server/streamableHttp.py

+
+    def _check_content_type(self, request: Request) -> bool:
+        """Check if the request has the correct Content-Type."""
+        content_type = request.headers.get("content-type", "")


Are request.headers always lowercase?

dsp-ant · 2025-04-30T11:29:07Z

src/mcp/server/streamableHttp.py

+        content_type_parts = [
+            part.strip() for part in content_type.split(";")[0].split(",")
+        ]
+
+        return any(part == CONTENT_TYPE_JSON for part in content_type_parts)


Suggested change

content_type_parts = [

part.strip() for part in content_type.split(";")[0].split(",")

]

return any(part == CONTENT_TYPE_JSON for part in content_type_parts)

content_type_parts = (

part.strip() for part in content_type.split(";")[0].split(",")

)

return any(part == CONTENT_TYPE_JSON for part in content_type_parts)

We never use content_type_parts other than checking it, we either can traverse it in a for-loop or should just use a generator here to be lazy about it.

dsp-ant · 2025-04-30T11:32:14Z

src/mcp/server/streamableHttp.py

+    async def _handle_post_request(
+        self, scope: Scope, request: Request, receive: Receive, send: Send
+    ) -> None:


Can we break this up? I think some of the branches, particularly the one where we handle an existing session and a non-existing session should probably be separate functions.

dsp-ant · 2025-04-30T13:28:12Z

src/mcp/server/streamableHttp.py

+                    if request_session_id and request_session_id != self.mcp_session_id:
+                        response = self._create_error_response(
+                            "Not Found: Invalid or expired session ID",
+                            HTTPStatus.NOT_FOUND,
+                        )


Are we sure this is 404 or maybe rather a 400 BAD REQUEST? What does the TS SDK do here?

dsp-ant

I'll accept this for now under the assumption that the fixes are done on top.

ihrpr added 13 commits April 21, 2025 09:36

initial streamable http server

2b95598

add request validation and tests

3d790f8

session management

27bc01e

terminations of a session

3c4cf10

fix cleaning up

bce74b3

add happy path test

2011579

tests

2cebf08

json mode

6c9c320

clean up

ede8cde

fix example server

2a3bed8

return 405 for get stream

0456b1b

speed up tests

97ca48d

format

92d4287

ihrpr added this to 2025-03-26 Implementation Apr 21, 2025

ihrpr requested review from dsp-ant, jspahrsummers and jerome3o-anthropic April 21, 2025 16:20

uv lock

aa9f6e5

ihrpr mentioned this pull request Apr 21, 2025

StreamableHttp - stateless server support #554

Merged

RamAnanth mentioned this pull request Apr 21, 2025

Server implementation of Streamable HTTP #461

Closed

8 tasks

This was referenced Apr 21, 2025

Implement Streamable HTTP transport #443

Open

Streamable HTTP Support #405

Closed

clean up

46ec72d

jerome3o-anthropic requested changes Apr 23, 2025

View reviewed changes

github-project-automation bot moved this to In Progress in 2025-03-26 Implementation Apr 23, 2025

ihrpr added 4 commits April 23, 2025 09:44

add comments to server example where we use related_request_id

9b096dc

Merge branch 'main' into ihrpr/streamablehttp-server

bbe79c2

small fixes

a0a9c5b

use mta field for related_request_id

a5ac2e0

unrelated test and format

2e615f3

ihrpr requested a review from jerome3o-anthropic April 23, 2025 10:11

ihrpr mentioned this pull request Apr 27, 2025

use metadata from SessionMessage to propagate related_request_id #591

Merged

jerome3o-anthropic approved these changes Apr 29, 2025

View reviewed changes

ihrpr added this to the 2025-03-26 spec release milestone Apr 29, 2025

vitorbal mentioned this pull request Apr 30, 2025

Streamable HTTP support jlowin/fastmcp#130

Open

dsp-ant requested changes Apr 30, 2025

View reviewed changes

smurching mentioned this pull request May 1, 2025

[WIP] Prototype/POC of remote server support + hosting on Databricks apps databrickslabs/mcp#22

Open

12 tasks

dsp-ant self-requested a review May 1, 2025 21:46

dsp-ant approved these changes May 1, 2025

View reviewed changes

Merge branch 'main' into ihrpr/streamablehttp-server

ff70bd6

ihrpr merged commit 78f0b11 into main May 2, 2025
11 checks passed

github-project-automation bot moved this from In Progress to Done in 2025-03-26 Implementation May 2, 2025

ihrpr deleted the ihrpr/streamablehttp-server branch May 2, 2025 10:58

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

StreamableHttp - Server transport with state management #553

StreamableHttp - Server transport with state management #553

ihrpr commented Apr 21, 2025

jerome3o-anthropic left a comment

jerome3o-anthropic Apr 23, 2025

jerome3o-anthropic Apr 23, 2025

jerome3o-anthropic Apr 23, 2025

jerome3o-anthropic Apr 23, 2025

ihrpr Apr 23, 2025

ihrpr commented Apr 23, 2025 •

edited

Loading

ihrpr commented Apr 27, 2025

jerome3o-anthropic left a comment

dsp-ant Apr 30, 2025

dsp-ant Apr 30, 2025

dsp-ant Apr 30, 2025

ihrpr Apr 30, 2025

dsp-ant Apr 30, 2025

ihrpr Apr 30, 2025

dsp-ant Apr 30, 2025

dsp-ant Apr 30, 2025

dsp-ant Apr 30, 2025

dsp-ant Apr 30, 2025

dsp-ant Apr 30, 2025

dsp-ant left a comment

StreamableHttp - Server transport with state management #553

StreamableHttp - Server transport with state management #553

Conversation

ihrpr commented Apr 21, 2025

Features

Follow ups

jerome3o-anthropic left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ihrpr commented Apr 23, 2025 • edited Loading

ihrpr commented Apr 27, 2025

jerome3o-anthropic left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dsp-ant left a comment

Choose a reason for hiding this comment

ihrpr commented Apr 23, 2025 •

edited

Loading