StreamableHttp client transport

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

@@ -7,7 +7,7 @@
 import click
 import mcp.types as types
 from mcp.server.lowlevel import Server
-from mcp.server.streamableHttp import (
+from mcp.server.streamable_http import (
     MCP_SESSION_ID_HEADER,
     StreamableHTTPServerTransport,
 )

src/mcp/client/streamable_http.py

@@ -0,0 +1,258 @@
+"""
+StreamableHTTP Client Transport Module
+
+This module implements the StreamableHTTP transport for MCP clients,
+providing support for HTTP POST requests with optional SSE streaming responses
+and session management.
+"""
+
+import logging
+from contextlib import asynccontextmanager
+from datetime import timedelta
+from typing import Any
+
+import anyio
+import httpx
+from httpx_sse import EventSource, aconnect_sse
+
+from mcp.types import (
+    ErrorData,
+    JSONRPCError,
+    JSONRPCMessage,
+    JSONRPCNotification,
+    JSONRPCRequest,
+)
+
+logger = logging.getLogger(__name__)
+
+# Header names
+MCP_SESSION_ID_HEADER = "mcp-session-id"
+LAST_EVENT_ID_HEADER = "last-event-id"
+
+# Content types
+CONTENT_TYPE_JSON = "application/json"
+CONTENT_TYPE_SSE = "text/event-stream"
+
+
+@asynccontextmanager
+async def streamablehttp_client(
+    url: str,
+    headers: dict[str, Any] | None = None,
+    timeout: timedelta = timedelta(seconds=30),
+    sse_read_timeout: timedelta = timedelta(seconds=60 * 5),
+):
+    """
+    Client transport for StreamableHTTP.
+
+    `sse_read_timeout` determines how long (in seconds) the client will wait for a new
+    event before disconnecting. All other HTTP operations are controlled by `timeout`.
+
+    Yields:
+        Tuple of (read_stream, write_stream, terminate_callback)
+    """
+
+    read_stream_writer, read_stream = anyio.create_memory_object_stream[
+        JSONRPCMessage | Exception
+    ](0)
+    write_stream, write_stream_reader = anyio.create_memory_object_stream[
+        JSONRPCMessage
+    ](0)
+
+    async def get_stream():
+        """
+        Optional GET stream for server-initiated messages
+        """
+        nonlocal session_id
+        try:
+            # Only attempt GET if we have a session ID
+            if not session_id:
+                return
+
+            get_headers = request_headers.copy()
+            get_headers[MCP_SESSION_ID_HEADER] = session_id
+
+            async with aconnect_sse(
+                client,
+                "GET",
+                url,
+                headers=get_headers,
+                timeout=httpx.Timeout(timeout.seconds, read=sse_read_timeout.seconds),
+            ) as event_source:
+                event_source.response.raise_for_status()
+                logger.debug("GET SSE connection established")
+
+                async for sse in event_source.aiter_sse():
+                    if sse.event == "message":
+                        try:
+                            message = JSONRPCMessage.model_validate_json(sse.data)
+                            logger.debug(f"GET message: {message}")
+                            await read_stream_writer.send(message)
+                        except Exception as exc:
+                            logger.error(f"Error parsing GET message: {exc}")
+                            await read_stream_writer.send(exc)
+                    else:
+                        logger.warning(f"Unknown SSE event from GET: {sse.event}")
+        except Exception as exc:
+            # GET stream is optional, so don't propagate errors
+            logger.debug(f"GET stream error (non-fatal): {exc}")
+
+    async def post_writer(client: httpx.AsyncClient):
+        nonlocal session_id
+        try:
+            async with write_stream_reader:
+                async for message in write_stream_reader:
+                    # Add session ID to headers if we have one
+                    post_headers = request_headers.copy()
+                    if session_id:
+                        post_headers[MCP_SESSION_ID_HEADER] = session_id
+
+                    logger.debug(f"Sending client message: {message}")
+
+                    # Handle initial initialization request
+                    is_initialization = (
+                        isinstance(message.root, JSONRPCRequest)
+                        and message.root.method == "initialize"
+                    )
+                    if (
+                        isinstance(message.root, JSONRPCNotification)
+                        and message.root.method == "notifications/initialized"
+                    ):
+                        tg.start_soon(get_stream)
+
+                    async with client.stream(
+                        "POST",
+                        url,
+                        json=message.model_dump(
+                            by_alias=True, mode="json", exclude_none=True
+                        ),
+                        headers=post_headers,
+                    ) as response:
+                        if response.status_code == 202:
+                            logger.debug("Received 202 Accepted")
+                            continue
+                        # Check for 404 (session expired/invalid)
+                        if response.status_code == 404:
+                            if isinstance(message.root, JSONRPCRequest):
+                                jsonrpc_error = JSONRPCError(
+                                    jsonrpc="2.0",
+                                    id=message.root.id,
+                                    error=ErrorData(
+                                        code=32600,
+                                        message="Session terminated",
+                                    ),
+                                )
+                                await read_stream_writer.send(
+                                    JSONRPCMessage(jsonrpc_error)
+                                )
+                                continue
+                        response.raise_for_status()
+
+                        # Extract session ID from response headers
+                        if is_initialization:
+                            new_session_id = response.headers.get(MCP_SESSION_ID_HEADER)
+                            if new_session_id:
+                                session_id = new_session_id
+                                logger.info(f"Received session ID: {session_id}")
+
+                        # Handle different response types
+                        content_type = response.headers.get("content-type", "").lower()
+
+                        if content_type.startswith(CONTENT_TYPE_JSON):
+                            try:
+                                content = await response.aread()
+                                json_message = JSONRPCMessage.model_validate_json(
+                                    content
+                                )
+                                await read_stream_writer.send(json_message)
+                            except Exception as exc:
+                                logger.error(f"Error parsing JSON response: {exc}")
+                                await read_stream_writer.send(exc)
+
+                        elif content_type.startswith(CONTENT_TYPE_SSE):
+                            # Parse SSE events from the response
+                            try:
+                                event_source = EventSource(response)
+                                async for sse in event_source.aiter_sse():
+                                    if sse.event == "message":
+                                        try:
+                                            await read_stream_writer.send(
+                                                JSONRPCMessage.model_validate_json(
+                                                    sse.data
+                                                )
+                                            )
+                                        except Exception as exc:
+                                            logger.exception("Error parsing message")
+                                            await read_stream_writer.send(exc)
+                                    else:
+                                        logger.warning(f"Unknown event: {sse.event}")
+
+                            except Exception as e:
+                                logger.exception("Error reading SSE stream:")
+                                await read_stream_writer.send(e)
+
+                        else:
+                            # For 202 Accepted with no body
+                            if response.status_code == 202:
+                                logger.debug("Received 202 Accepted")
+                                continue
+
+                            error_msg = f"Unexpected content type: {content_type}"
+                            logger.error(error_msg)
+                            await read_stream_writer.send(ValueError(error_msg))
+
+        except Exception as exc:
+            logger.error(f"Error in post_writer: {exc}")
+        finally:
+            await read_stream_writer.aclose()
+            await write_stream.aclose()
+
+    async def terminate_session():
+        """
+        Terminate the session by sending a DELETE request.
+        """
+        nonlocal session_id
+        if not session_id:
+            return  # No session to terminate
+
+        try:
+            delete_headers = request_headers.copy()
+            delete_headers[MCP_SESSION_ID_HEADER] = session_id
+
+            response = await client.delete(
+                url,
+                headers=delete_headers,
+            )
+
+            if response.status_code == 405:
+                # Server doesn't allow client-initiated termination
+                logger.debug("Server does not allow session termination")
+            elif response.status_code != 200:
+                logger.warning(f"Session termination failed: {response.status_code}")
+        except Exception as exc:
+            logger.warning(f"Session termination failed: {exc}")
+
+    async with anyio.create_task_group() as tg:
+        try:
+            logger.info(f"Connecting to StreamableHTTP endpoint: {url}")
+            # Set up headers with required Accept header
+            request_headers = {
+                "Accept": f"{CONTENT_TYPE_JSON}, {CONTENT_TYPE_SSE}",
+                "Content-Type": CONTENT_TYPE_JSON,
+                **(headers or {}),
+            }
+            # Track session ID if provided by server
+            session_id: str | None = None
+
+            async with httpx.AsyncClient(
+                headers=request_headers,
+                timeout=httpx.Timeout(timeout.seconds, read=sse_read_timeout.seconds),
+                follow_redirects=True,
+            ) as client:
+                tg.start_soon(post_writer, client)
+                try:
+                    yield read_stream, write_stream, terminate_session
+                finally:
+                    tg.cancel_scope.cancel()
+        finally:
+            await read_stream_writer.aclose()
+            await write_stream.aclose()