content: """
  Tests for the StreamableHTTP server and client transport.
  
  Contains tests for both server and client sides of the StreamableHTTP transport.
  """
  
  import json
  import multiprocessing
  import socket
  import time
  from collections.abc import Generator
  from typing import Any
  from unittest.mock import MagicMock
  
  import anyio
  import httpx
  import pytest
  import requests
  import uvicorn
  from httpx_sse import ServerSentEvent
  from pydantic import AnyUrl
  from starlette.applications import Starlette
  from starlette.requests import Request
  from starlette.routing import Mount
  
  import mcp.types as types
  from mcp.client.session import ClientSession
  from mcp.client.streamable_http import (
      StreamableHTTPTransport,
      streamable_http_client,
      streamablehttp_client,  # pyright: ignore[reportDeprecated]
  )
  from mcp.server import Server
  from mcp.server.streamable_http import (
      MCP_PROTOCOL_VERSION_HEADER,
      MCP_SESSION_ID_HEADER,
      SESSION_ID_PATTERN,
      EventCallback,
      EventId,
      EventMessage,
      EventStore,
      StreamableHTTPServerTransport,
      StreamId,
  )
  from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
  from mcp.server.transport_security import TransportSecuritySettings
  from mcp.shared._httpx_utils import create_mcp_http_client
  from mcp.shared.context import RequestContext
  from mcp.shared.exceptions import McpError
  from mcp.shared.message import ClientMessageMetadata, ServerMessageMetadata, SessionMessage
  from mcp.shared.session import RequestResponder
  from mcp.types import (
      InitializeResult,
      JSONRPCMessage,
      JSONRPCRequest,
      TextContent,
      TextResourceContents,
      Tool,
  )
  from tests.test_helpers import wait_for_server
  
  # Test constants
  SERVER_NAME = "test_streamable_http_server"
  TEST_SESSION_ID = "test-session-id-12345"
  INIT_REQUEST = {
      "jsonrpc": "2.0",
      "method": "initialize",
      "params": {
          "clientInfo": {"name": "test-client", "version": "1.0"},
          "protocolVersion": "2025-03-26",
          "capabilities": {},
      },
      "id": "init-1",
  }
  
  
  # Helper functions
  def extract_protocol_version_from_sse(response: requests.Response) -> str:  # pragma: no cover
      """Extract the negotiated protocol version from an SSE initialization response."""
      assert response.headers.get("Content-Type") == "text/event-stream"
      for line in response.text.splitlines():
          if line.startswith("data: "):
              init_data = json.loads(line[6:])
              return init_data["result"]["protocolVersion"]
      raise ValueError("Could not extract protocol version from SSE response")
  
  
  # Simple in-memory event store for testing
  class SimpleEventStore(EventStore):
      """Simple in-memory event store for testing."""
  
      def __init__(self):
          self._events: list[tuple[StreamId, EventId, types.JSONRPCMessage | None]] = []
          self._event_id_counter = 0
  
      async def store_event(  # pragma: no cover
          self, stream_id: StreamId, message: types.JSONRPCMessage | None
      ) -> EventId:
          """Store an event and return its ID."""
          self._event_id_counter += 1
          event_id = str(self._event_id_counter)
          self._events.append((stream_id, event_id, message))
          return event_id
  
      async def replay_events_after(  # pragma: no cover
          self,
          last_event_id: EventId,
          send_callback: EventCallback,
      ) -> StreamId | None:
          """Replay events after the specified ID."""
          # Find the stream ID of the last event
          target_stream_id = None
          for stream_id, event_id, _ in self._events:
              if event_id == last_event_id:
                  target_stream_id = stream_id
                  break
  
          if target_stream_id is None:
              # If event ID not found, return None
              return None
  
          # Convert last_event_id to int for comparison
          last_event_id_int = int(last_event_id)
  
          # Replay only events from the same stream with ID > last_event_id
          for stream_id, event_id, message in self._events:
              if stream_id == target_stream_id and int(event_id) > last_event_id_int:
                  # Skip priming events (None message)
                  if message is not None:
                      await send_callback(EventMessage(message, event_id))
  
          return target_stream_id
  
  
  # Test server implementation that follows MCP protocol
  class ServerTest(Server):  # pragma: no cover
      def __init__(self):
          super().__init__(SERVER_NAME)
          self._lock = None  # Will be initialized in async context
  
          @self.read_resource()
          async def handle_read_resource(uri: AnyUrl) -> str | bytes:
              if uri.scheme == "foobar":
                  return f"Read {uri.host}"
              elif uri.scheme == "slow":
                  # Simulate a slow resource
                  await anyio.sleep(2.0)
                  return f"Slow response from {uri.host}"
  
              raise ValueError(f"Unknown resource: {uri}")
  
          @self.list_tools()
          async def handle_list_tools() -> list[Tool]:
              return [
                  Tool(
                      name="test_tool",
                      description="A test tool",
                      inputSchema={"type": "object", "properties": {}},
                  ),
                  Tool(
                      name="test_tool_with_standalone_notification",
                      description="A test tool that sends a notification",
                      inputSchema={"type": "object", "properties": {}},
                  ),
                  Tool(
                      name="long_running_with_checkpoints",
                      description="A long-running tool that sends periodic notifications",
                      inputSchema={"type": "object", "properties": {}},
                  ),
                  Tool(
                      name="test_sampling_tool",
                      description="A tool that triggers server-side sampling",
                      inputSchema={"type": "object", "properties": {}},
                  ),
                  Tool(
                      name="wait_for_lock_with_notification",
                      description="A tool that sends a notification and waits for lock",
                      inputSchema={"type": "object", "properties": {}},
                  ),
                  Tool(
                      name="release_lock",
                      description="A tool that releases the lock",
                      inputSchema={"type": "object", "properties": {}},
                  ),
                  Tool(
                      name="tool_with_stream_close",
                      description="A tool that closes SSE stream mid-operation",
                      inputSchema={"type": "object", "properties": {}},
                  ),
                  Tool(
                      name="tool_with_multiple_notifications_and_close",
                      description="Tool that sends notification1, closes stream, sends notification2, notification3",
                      inputSchema={"type": "object", "properties": {}},
                  ),
                  Tool(
                      name="tool_with_multiple_stream_closes",
                      description="Tool that closes SSE stream multiple times during execution",
                      inputSchema={
                          "type": "object",
                          "properties": {
                              "checkpoints": {"type": "integer", "default": 3},
                              "sleep_time": {"type": "number", "default": 0.2},
                          },
                      },
                  ),
                  Tool(
                      name="tool_with_standalone_stream_close",
                      description="Tool that closes standalone GET stream mid-operation",
                      inputSchema={"type": "object", "properties": {}},
                  ),
              ]
  
          @self.call_tool()
          async def handle_call_tool(name: str, args: dict[str, Any]) -> list[TextContent]:
              ctx = self.request_context
  
              # When the tool is called, send a notification to test GET stream
              if name == "test_tool_with_standalone_notification":
                  await ctx.session.send_resource_updated(uri=AnyUrl("http://test_resource"))
                  return [TextContent(type="text", text=f"Called {name}")]
  
              elif name == "long_running_with_checkpoints":
                  # Send notifications that are part of the response stream
                  # This simulates a long-running tool that sends logs
  
                  await ctx.session.send_log_message(
                      level="info",
                      data="Tool started",
                      logger="tool",
                      related_request_id=ctx.request_id,  # need for stream association
                  )
  
                  await anyio.sleep(0.1)
  
                  await ctx.session.send_log_message(
                      level="info",
                      data="Tool is almost done",
                      logger="tool",
                      related_request_id=ctx.request_id,
                  )
  
                  return [TextContent(type="text", text="Completed!")]
  
              elif name == "test_sampling_tool":
                  # Test sampling by requesting the client to sample a message
                  sampling_result = await ctx.session.create_message(
                      messages=[
                          types.SamplingMessage(
                              role="user",
                              content=types.TextContent(type="text", text="Server needs client sampling"),
                          )
                      ],
                      max_tokens=100,
                      related_request_id=ctx.request_id,
                  )
  
                  # Return the sampling result in the tool response
                  # Since we're not passing tools param, result.content is single content
                  if sampling_result.content.type == "text":
                      response = sampling_result.content.text
                  else:
                      response = str(sampling_result.content)
                  return [
                      TextContent(
                          type="text",
                          text=f"Response from sampling: {response}",
                      )
                  ]
  
              elif name == "wait_for_lock_with_notification":
                  # Initialize lock if not already done
                  if self._lock is None:
                      self._lock = anyio.Event()
  
                  # First send a notification
                  await ctx.session.send_log_message(
                      level="info",
                      data="First notification before lock",
                      logger="lock_tool",
                      related_request_id=ctx.request_id,
                  )
  
                  # Now wait for the lock to be released
                  await self._lock.wait()
  
                  # Send second notification after lock is released
                  await ctx.session.send_log_message(
                      level="info",
                      data="Second notification after lock",
                      logger="lock_tool",
                      related_request_id=ctx.request_id,
                  )
  
                  return [TextContent(type="text", text="Completed")]
  
              elif name == "release_lock":
                  assert self._lock is not None, "Lock must be initialized before releasing"
  
                  # Release the lock
                  self._lock.set()
                  return [TextContent(type="text", text="Lock released")]
  
              elif name == "tool_with_stream_close":
                  # Send notification before closing
                  await ctx.session.send_log_message(
                      level="info",
                      data="Before close",
                      logger="stream_close_tool",
                      related_request_id=ctx.request_id,
                  )
                  # Close SSE stream (triggers client reconnect)
                  assert ctx.close_sse_stream is not None
                  await ctx.close_sse_stream()
                  # Continue processing (events stored in event_store)
                  await anyio.sleep(0.1)
                  await ctx.session.send_log_message(
                      level="info",
                      data="After close",
                      logger="stream_close_tool",
                      related_request_id=ctx.request_id,
                  )
                  return [TextContent(type="text", text="Done")]
  
              elif name == "tool_with_multiple_notifications_and_close":
                  # Send notification1
                  await ctx.session.send_log_message(
                      level="info",
                      data="notification1",
                      logger="multi_notif_tool",
                      related_request_id=ctx.request_id,
                  )
                  # Close SSE stream
                  assert ctx.close_sse_stream is not None
                  await ctx.close_sse_stream()
                  # Send notification2, notification3 (stored in event_store)
                  await anyio.sleep(0.1)
                  await ctx.session.send_log_message(
                      level="info",
                      data="notification2",
                      logger="multi_notif_tool",
                      related_request_id=ctx.request_id,
                  )
                  await ctx.session.send_log_message(
                      level="info",
                      data="notification3",
                      logger="multi_notif_tool",
                      related_request_id=ctx.request_id,
                  )
                  return [TextContent(type="text", text="All notifications sent")]
  
              elif name == "tool_with_multiple_stream_closes":
                  num_checkpoints = args.get("checkpoints", 3)
                  sleep_time = args.get("sleep_time", 0.2)
  
                  for i in range(num_checkpoints):
                      await ctx.session.send_log_message(
                          level="info",
                          data=f"checkpoint_{i}",
                          logger="multi_close_tool",
                          related_request_id=ctx.request_id,
                      )
  
                      if ctx.close_sse_stream:
                          await ctx.close_sse_stream()
  
                      await anyio.sleep(sleep_time)
  
                  return [TextContent(type="text", text=f"Completed {num_checkpoints} checkpoints")]
  
              elif name == "tool_with_standalone_stream_close":
                  # Test for GET stream reconnection
                  # 1. Send unsolicited notification via GET stream (no related_request_id)
                  await ctx.session.send_resource_updated(uri=AnyUrl("http://notification_1"))
  
                  # Small delay to ensure notification is flushed before closing
                  await anyio.sleep(0.1)
  
                  # 2. Close the standalone GET stream
                  if ctx.close_standalone_sse_stream:
                      await ctx.close_standalone_sse_stream()
  
                  # 3. Wait for client to reconnect (uses retry_interval from server, default 1000ms)
                  await anyio.sleep(1.5)
  
                  # 4. Send another notification on the new GET stream connection
                  await ctx.session.send_resource_updated(uri=AnyUrl("http://notification_2"))
  
                  return [TextContent(type="text", text="Standalone stream close test done")]
  
              return [TextContent(type="text", text=f"Called {name}")]
  
  
  def create_app(
      is_json_response_enabled: bool = False,
      event_store: EventStore | None = None,
      retry_interval: int | None = None,
  ) -> Starlette:  # pragma: no cover
      """Create a Starlette application for testing using the session manager.
  
      Args:
          is_json_response_enabled: If True, use JSON responses instead of SSE streams.
          event_store: Optional event store for testing resumability.
          retry_interval: Retry interval in milliseconds for SSE polling.
      """
      # Create server instance
      server = ServerTest()
  
      # Create the session manager
      security_settings = TransportSecuritySettings(
          allowed_hosts=["127.0.0.1:*", "localhost:*"], allowed_origins=["http://127.0.0.1:*", "http://localhost:*"]
      )
      session_manager = StreamableHTTPSessionManager(
          app=server,
          event_store=event_store,
          json_response=is_json_response_enabled,
          security_settings=security_settings,
          retry_interval=retry_interval,
      )
  
      # Create an ASGI application that uses the session manager
      app = Starlette(
          debug=True,
          routes=[
              Mount("/mcp", app=session_manager.handle_request),
          ],
          lifespan=lambda app: session_manager.run(),
      )
  
      return app
  
  
  def run_server(
      port: int,
      is_json_response_enabled: bool = False,
      event_store: EventStore | None = None,
      retry_interval: int | None = None,
  ) -> None:  # pragma: no cover
      """Run the test server.
  
      Args:
          port: Port to listen on.
          is_json_response_enabled: If True, use JSON responses instead of SSE streams.
          event_store: Optional event store for testing resumability.
          retry_interval: Retry interval in milliseconds for SSE polling.
      """
  
      app = create_app(is_json_response_enabled, event_store, retry_interval)
      # Configure server
      config = uvicorn.Config(
          app=app,
          host="127.0.0.1",
          port=port,
          log_level="info",
          limit_concurrency=10,
          timeout_keep_alive=5,
          access_log=False,
      )
  
      # Start the server
      server = uvicorn.Server(config=config)
  
      # This is important to catch exceptions and prevent test hangs
      try:
          server.run()
      except Exception:
          import traceback
  
          traceback.print_exc()
  
  
  # Test fixtures - using same approach as SSE tests
  @pytest.fixture
  def basic_server_port() -> int:
      """Find an available port for the basic server."""
      with socket.socket() as s:
          s.bind(("127.0.0.1", 0))
          return s.getsockname()[1]
  
  
  @pytest.fixture
  def json_server_port() -> int:
      """Find an available port for the JSON response server."""
      with socket.socket() as s:
          s.bind(("127.0.0.1", 0))
          return s.getsockname()[1]
  
  
  @pytest.fixture
  def basic_server(basic_server_port: int) -> Generator[None, None, None]:
      """Start a basic server."""
      proc = multiprocessing.Process(target=run_server, kwargs={"port": basic_server_port}, daemon=True)
      proc.start()
  
      # Wait for server to be running
      wait_for_server(basic_server_port)
  
      yield
  
      # Clean up
      proc.kill()
      proc.join(timeout=2)
  
  
  @pytest.fixture
  def event_store() -> SimpleEventStore:
      """Create a test event store."""
      return SimpleEventStore()
  
  
  @pytest.fixture
  def event_server_port() -> int:
      """Find an available port for the event store server."""
      with socket.socket() as s:
          s.bind(("127.0.0.1", 0))
          return s.getsockname()[1]
  
  
  @pytest.fixture
  def event_server(
      event_server_port: int, event_store: SimpleEventStore
  ) -> Generator[tuple[SimpleEventStore, str], None, None]:
      """Start a server with event store and retry_interval enabled."""
      proc = multiprocessing.Process(
          target=run_server,
          kwargs={"port": event_server_port, "event_store": event_store, "retry_interval": 500},
          daemon=True,
      )
      proc.start()
  
      # Wait for server to be running
      wait_for_server(event_server_port)
  
      yield event_store, f"http://127.0.0.1:{event_server_port}"
  
      # Clean up
      proc.kill()
      proc.join(timeout=2)
  
  
  @pytest.fixture
  def json_response_server(json_server_port: int) -> Generator[None, None, None]:
      """Start a server with JSON response enabled."""
      proc = multiprocessing.Process(
          target=run_server,
          kwargs={"port": json_server_port, "is_json_response_enabled": True},
          daemon=True,
      )
      proc.start()
  
      # Wait for server to be running
      wait_for_server(json_server_port)
  
      yield
  
      # Clean up
      proc.kill()
      proc.join(timeout=2)
  
  
  @pytest.fixture
  def basic_server_url(basic_server_port: int) -> str:
      """Get the URL for the basic test server."""
      return f"http://127.0.0.1:{basic_server_port}"
  
  
  @pytest.fixture
  def json_server_url(json_server_port: int) -> str:
      """Get the URL for the JSON response test server."""
      return f"http://127.0.0.1:{json_server_port}"
  
  
  # Basic request validation tests
  def test_accept_header_validation(basic_server: None, basic_server_url: str):
      """Test that Accept header is properly validated."""
      # Test without Accept header
      response = requests.post(
          f"{basic_server_url}/mcp",
          headers={"Content-Type": "application/json"},
          json={"jsonrpc": "2.0", "method": "initialize", "id": 1},
      )
      assert response.status_code == 406
      assert "Not Acceptable" in response.text
  
  
  def test_content_type_validation(basic_server: None, basic_server_url: str):
      """Test that Content-Type header is properly validated."""
      # Test with incorrect Content-Type
      response = requests.post(
          f"{basic_server_url}/mcp",
          headers={
              "Accept": "application/json, text/event-stream",
              "Content-Type": "text/plain",
          },
          data="This is not JSON",
      )
  
      assert response.status_code == 400
      assert "Invalid Content-Type" in response.text
  
  
  def test_json_validation(basic_server: None, basic_server_url: str):
      """Test that JSON content is properly validated."""
      # Test with invalid JSON
      response = requests.post(
          f"{basic_server_url}/mcp",
          headers={
              "Accept": "application/json, text/event-stream",
              "Content-Type": "application/json",
          },
          data="this is not valid json",
      )
      assert response.status_code == 400
      assert "Parse error" in response.text
  
  
  def test_json_parsing(basic_server: None, basic_server_url: str):
      """Test that JSON content is properly parse."""
      # Test with valid JSON but invalid JSON-RPC
      response = requests.post(
          f"{basic_server_url}/mcp",
          headers={
              "Accept": "application/json, text/event-stream",
              "Content-Type": "application/json",
          },
          json={"foo": "bar"},
      )
      assert response.status_code == 400
      assert "Validation error" in response.text
  
  
  def test_method_not_allowed(basic_server: None, basic_server_url: str):
      """Test that unsupported HTTP methods are rejected."""
      # Test with unsupported method (PUT)
      response = requests.put(
          f"{basic_server_url}/mcp",
          headers={
              "Accept": "application/json, text/event-stream",
              "Content-Type": "application/json",
          },
          json={"jsonrpc": "2.0", "method": "initialize", "id": 1},
      )
      assert response.status_code == 405
      assert "Method Not Allowed" in response.text
  
  
  def test_session_validation(basic_server: None, basic_server_url: str):
      """Test session ID validation."""
      # session_id not used directly in this test
  
      # Test without session ID
      response = requests.post(
          f"{basic_server_url}/mcp",
          headers={
              "Accept": "application/json, text/event-stream",
              "Content-Type": "application/json",
          },
          json={"jsonrpc": "2.0", "method": "list_tools", "id": 1},
      )
      assert response.status_code == 400
      assert "Missing session ID" in response.text
  
  
  def test_session_id_pattern():
      """Test that SESSION_ID_PATTERN correctly validates session IDs."""
      # Valid session IDs (visible ASCII characters from 0x21 to 0x7E)
      valid_session_ids = [
          "test-session-id",
          "1234567890",
          "session!@#$%^&*()_+-=[]{}|;:,.<>?/",
          "~`",
      ]
  
      for session_id in valid_session_ids:
          assert SESSION_ID_PATTERN.match(session_id) is not None
          # Ensure fullmatch matches too (whole string)
          assert SESSION_ID_PATTERN.fullmatch(session_id) is not None
  
      # Invalid session IDs
      invalid_session_ids = [
          "",  # Empty string
          " test",  # Space (0x20)
          "test\t",  # Tab
          "test\n",  # Newline
          "test\r",  # Carriage return
          "test" + chr(0x7F),  # DEL character
          "test" + chr(0x80),  # Extended ASCII
          "test" + chr(0x00),  # Null character
          "test" + chr(0x20),  # Space (0x20)
      ]
  
      for session_id in invalid_session_ids:
          # For invalid IDs, either match will fail or fullmatch will fail
          if SESSION_ID_PATTERN.match(session_id) is not None:
              # If match succeeds, fullmatch should fail (partial match case)
              assert SESSION_ID_PATTERN.fullmatch(session_id) is None
  
  
  def test_streamable_http_transport_init_validation():
      """Test that StreamableHTTPServerTransport validates session ID on init."""
      # Valid session ID should initialize without errors
      valid_transport = StreamableHTTPServerTransport(mcp_session_id="valid-id")
      assert valid_transport.mcp_session_id == "valid-id"
  
      # None should be accepted
      none_transport = StreamableHTTPServerTransport(mcp_session_id=None)
      assert none_transport.mcp_session_id is None
  
      # Invalid session ID should raise ValueError
      with pytest.raises(ValueError) as excinfo:
          StreamableHTTPServerTransport(mcp_session_id="invalid id with space")
      assert "Session ID must only contain visible ASCII characters" in str(excinfo.value)
  
      # Test with control characters
      with pytest.raises(ValueError):
          StreamableHTTPServerTransport(mcp_session_id="test\nid")
  
      with pytest.raises(ValueError):
          StreamableHTTPServerTransport(mcp_session_id="test\n")
  
  
  def test_session_termination(basic_server: None, basic_server_url: str):
      """Test session termination via DELETE and subsequent request handling."""
      response = requests.post(
          f"{basic_server_url}/mcp",
          headers={
              "Accept": "application/json, text/event-stream",
              "Content-Type": "application/json",
          },
          json=INIT_REQUEST,
      )
      assert response.status_code == 200
  
      # Extract negotiated protocol version from SSE response
      negotiated_version = extract_protocol_version_from_sse(response)
  
      # Now terminate the session
      session_id = response.headers.get(MCP_SESSION_ID_HEADER)
      response = requests.delete(
          f"{basic_server_url}/mcp",
          headers={
              MCP_SESSION_ID_HEADER: session_id,
              MCP_PROTOCOL_VERSION_HEADER: negotiated_version,
          },
      )
      assert response.status_code == 200
  
      # Try to use the terminated session
      response = requests.post(
          f"{basic_server_url}/mcp",
          headers={
              "Accept": "application/json, text/event-stream",
              "Content-Type": "application/json",
              MCP_SESSION_ID_HEADER: session_id,
          },
          json={"jsonrpc": "2.0", "method": "ping", "id": 2},
      )
      assert response.status_code == 404
      assert "Session has been terminated" in response.text
  
  
  def test_response(basic_server: None, basic_server_url: str):
      """Test response handling for a valid request."""
      mcp_url = f"{basic_server_url}/mcp"
      response = requests.post(
          mcp_url,
          headers={
              "Accept": "application/json, text/event-stream",
              "Content-Type": "application/json",
          },
          json=INIT_REQUEST,
      )
      assert response.status_code == 200
  
      # Extract negotiated protocol version from SSE response
      negotiated_version = extract_protocol_version_from_sse(response)
  
      # Now get the session ID
      session_id = response.headers.get(MCP_SESSION_ID_HEADER)
  
      # Try to use the session with proper headers
      tools_response = requests.post(
          mcp_url,
          headers={
              "Accept": "application/json, text/event-stream",
              "Content-Type": "application/json",
              MCP_SESSION_ID_HEADER: session_id,  # Use the session ID we got earlier
              MCP_PROTOCOL_VERSION_HEADER: negotiated_version,
          },
          json={"jsonrpc": "2.0", "method": "tools/list", "id": "tools-1"},
          stream=True,
      )
      assert tools_response.status_code == 200
      assert tools_response.headers.get("Content-Type") == "text/event-stream"
  
  
  def test_json_response(json_response_server: None, json_server_url: str):
      """Test response handling when is_json_response_enabled is True."""
      mcp_url = f"{json_server_url}/mcp"
      response = requests.post(
          mcp_url,
          headers={
              "Accept": "application/json, text/event-stream",
              "Content-Type": "application/json",
          },
          json=INIT_REQUEST,
      )
      assert response.status_code == 200
      assert response.headers.get("Content-Type") == "application/json"
  
  
  def test_json_response_accept_json_only(json_response_server: None, json_server_url: str):
      """Test that json_response servers only require application/json in Accept header."""
      mcp_url = f"{json_server_url}/mcp"
      response = requests.post(
          mcp_url,
          headers={
              "Accept": "application/json",
              "Content-Type": "application/json",
          },
          json=INIT_REQUEST,
      )
      assert response.status_code == 200
      assert response.headers.get("Content-Type") == "application/json"
  
  
  def test_json_response_missing_accept_header(json_response_server: None, json_server_url: str):
      """Test that json_response servers reject requests without Accept header."""
      mcp_url = f"{json_server_url}/mcp"
      response = requests.post(
          mcp_url,
          headers={
              "Content-Type": "application/json",
          },
          json=INIT_REQUEST,
      )
      assert response.status_code == 406
      assert "Not Acceptable" in response.text
  
  
  def test_json_response_incorrect_accept_header(json_response_server: None, json_server_url: str):
      """Test that json_response servers reject requests with incorrect Accept header."""
      mcp_url = f"{json_server_url}/mcp"
      # Test with only text/event-stream (wrong for JSON server)
      response = requests.post(
          mcp_url,
          headers={
              "Accept": "text/event-stream",
              "Content-Type": "application/json",
          },
          json=INIT_REQUEST,
      )
      assert response.status_code == 406
      assert "Not Acceptable" in response.text
  
  
  def test_get_sse_stream(basic_server: None, basic_server_url: str):
      """Test establishing an SSE stream via GET request."""
      # First, we need to initialize a session
      mcp_url = f"{basic_server_url}/mcp"
      init_response = requests.post(
          mcp_url,
          headers={
              "Accept": "application/json, text/event-stream",
              "Content-Type": "application/json",
          },
          json=INIT_REQUEST,
      )
      assert init_response.status_code == 200
  
      # Get the session ID
      session_id = init_response.headers.get(MCP_SESSION_ID_HEADER)
      assert session_id is not None
  
      # Extract negotiated protocol version from SSE response
      init_data = None
      assert init_response.headers.get("Content-Type") == "text/event-stream"
      for line in init_response.text.splitlines():  # pragma: no branch
          if line.startswith("data: "):  # pragma: no cover
              init_data = json.loads(line[6:])
              break
      assert init_data is not None
      negotiated_version = init_data["result"]["protocolVersion"]
  
      # Now attempt to establish an SSE stream via GET
      get_response = requests.get(
          mcp_url,
          headers={
              "Accept": "text/event-stream",
              MCP_SESSION_ID_HEADER: session_id,
              MCP_PROTOCOL_VERSION_HEADER: negotiated_version,
          },
          stream=True,
      )
  
      # Verify we got a successful response with the right content type
      assert get_response.status_code == 200
      assert get_response.headers.get("Content-Type") == "text/event-stream"
  
      # Test that a second GET request gets rejected (only one stream allowed)
      second_get = requests.get(
          mcp_url,
          headers={
              "Accept": "text/event-stream",
              MCP_SESSION_ID_HEADER: session_id,
              MCP_PROTOCOL_VERSION_HEADER: negotiated_version,
          },
          stream=True,
      )
  
      # Should get CONFLICT (409) since there's already a stream
      # Note: This might fail if the first stream fully closed before this runs,
      # but generally it should work in the test environment where it runs quickly
      assert second_get.status_code == 409
  
  
  def test_get_validation(basic_server: None, basic_server_url: str):
      """Test validation for GET requests."""
      # First, we need to initialize a session
      mcp_url = f"{basic_server_url}/mcp"
      init_response = requests.post(
          mcp_url,
          headers={
  