content: """
  Tests for Unicode handling in streamable HTTP transport.
  
  Verifies that Unicode text is correctly transmitted and received in both directions
  (server→client and client→server) using the streamable HTTP transport.
  """
  
  import multiprocessing
  import socket
  from collections.abc import Generator
  
  import pytest
  
  from mcp.client.session import ClientSession
  from mcp.client.streamable_http import streamable_http_client
  from tests.test_helpers import wait_for_server
  
  # Test constants with various Unicode characters
  UNICODE_TEST_STRINGS = {
      "cyrillic": "Слой хранилища, где располагаются",
      "cyrillic_short": "Привет мир",
      "chinese": "你好世界 - 这是一个测试",
      "japanese": "こんにちは世界 - これはテストです",
      "korean": "안녕하세요 세계 - 이것은 테스트입니다",
      "arabic": "مرحبا بالعالم - هذا اختبار",
      "hebrew": "שלום עולם - זה מבחן",
      "greek": "Γεια σου κόσμε - αυτό είναι δοκιμή",
      "emoji": "Hello 👋 World 🌍 - Testing 🧪 Unicode ✨",
      "math": "∑ ∫ √ ∞ ≠ ≤ ≥ ∈ ∉ ⊆ ⊇",
      "accented": "Café, naïve, résumé, piñata, Zürich",
      "mixed": "Hello世界🌍Привет안녕مرحباשלום",
      "special": "Line\nbreak\ttab\r\nCRLF",
      "quotes": '«French» „German" "English" 「Japanese」',
      "currency": "€100 £50 ¥1000 ₹500 ₽200 ¢99",
  }
  
  
  def run_unicode_server(port: int) -> None:  # pragma: no cover
      """Run the Unicode test server in a separate process."""
      # Import inside the function since this runs in a separate process
      from collections.abc import AsyncGenerator
      from contextlib import asynccontextmanager
      from typing import Any
  
      import uvicorn
      from starlette.applications import Starlette
      from starlette.routing import Mount
  
      import mcp.types as types
      from mcp.server import Server
      from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
      from mcp.types import TextContent, Tool
  
      # Need to recreate the server setup in this process
      server = Server(name="unicode_test_server")
  
      @server.list_tools()
      async def list_tools() -> list[Tool]:
          """List tools with Unicode descriptions."""
          return [
              Tool(
                  name="echo_unicode",
                  description="🔤 Echo Unicode text - Hello 👋 World 🌍 - Testing 🧪 Unicode ✨",
                  inputSchema={
                      "type": "object",
                      "properties": {
                          "text": {"type": "string", "description": "Text to echo back"},
                      },
                      "required": ["text"],
                  },
              ),
          ]
  
      @server.call_tool()
      async def call_tool(name: str, arguments: dict[str, Any] | None) -> list[TextContent]:
          """Handle tool calls with Unicode content."""
          if name == "echo_unicode":
              text = arguments.get("text", "") if arguments else ""
              return [
                  TextContent(
                      type="text",
                      text=f"Echo: {text}",
                  )
              ]
          else:
              raise ValueError(f"Unknown tool: {name}")
  
      @server.list_prompts()
      async def list_prompts() -> list[types.Prompt]:
          """List prompts with Unicode names and descriptions."""
          return [
              types.Prompt(
                  name="unicode_prompt",
                  description="Unicode prompt - Слой хранилища, где располагаются",
                  arguments=[],
              )
          ]
  
      @server.get_prompt()
      async def get_prompt(name: str, arguments: dict[str, Any] | None) -> types.GetPromptResult:
          """Get a prompt with Unicode content."""
          if name == "unicode_prompt":
              return types.GetPromptResult(
                  messages=[
                      types.PromptMessage(
                          role="user",
                          content=types.TextContent(
                              type="text",
                              text="Hello世界🌍Привет안녕مرحباשלום",
                          ),
                      )
                  ]
              )
          raise ValueError(f"Unknown prompt: {name}")
  
      # Create the session manager
      session_manager = StreamableHTTPSessionManager(
          app=server,
          json_response=False,  # Use SSE for testing
      )
  
      @asynccontextmanager
      async def lifespan(app: Starlette) -> AsyncGenerator[None, None]:
          async with session_manager.run():
              yield
  
      # Create an ASGI application
      app = Starlette(
          debug=True,
          routes=[
              Mount("/mcp", app=session_manager.handle_request),
          ],
          lifespan=lifespan,
      )
  
      # Run the server
      config = uvicorn.Config(
          app=app,
          host="127.0.0.1",
          port=port,
          log_level="error",
      )
      uvicorn_server = uvicorn.Server(config)
      uvicorn_server.run()
  
  
  @pytest.fixture
  def unicode_server_port() -> int:
      """Find an available port for the Unicode test server."""
      with socket.socket() as s:
          s.bind(("127.0.0.1", 0))
          return s.getsockname()[1]
  
  
  @pytest.fixture
  def running_unicode_server(unicode_server_port: int) -> Generator[str, None, None]:
      """Start a Unicode test server in a separate process."""
      proc = multiprocessing.Process(target=run_unicode_server, kwargs={"port": unicode_server_port}, daemon=True)
      proc.start()
  
      # Wait for server to be ready
      wait_for_server(unicode_server_port)
  
      try:
          yield f"http://127.0.0.1:{unicode_server_port}"
      finally:
          # Clean up - try graceful termination first
          proc.terminate()
          proc.join(timeout=2)
          if proc.is_alive():  # pragma: no cover
              proc.kill()
              proc.join(timeout=1)
  
  
  @pytest.mark.anyio
  async def test_streamable_http_client_unicode_tool_call(running_unicode_server: str) -> None:
      """Test that Unicode text is correctly handled in tool calls via streamable HTTP."""
      base_url = running_unicode_server
      endpoint_url = f"{base_url}/mcp"
  
      async with streamable_http_client(endpoint_url) as (read_stream, write_stream, _get_session_id):
          async with ClientSession(read_stream, write_stream) as session:
              await session.initialize()
  
              # Test 1: List tools (server→client Unicode in descriptions)
              tools = await session.list_tools()
              assert len(tools.tools) == 1
  
              # Check Unicode in tool descriptions
              echo_tool = tools.tools[0]
              assert echo_tool.name == "echo_unicode"
              assert echo_tool.description is not None
              assert "🔤" in echo_tool.description
              assert "👋" in echo_tool.description
  
              # Test 2: Send Unicode text in tool call (client→server→client)
              for test_name, test_string in UNICODE_TEST_STRINGS.items():
                  result = await session.call_tool("echo_unicode", arguments={"text": test_string})
  
                  # Verify server correctly received and echoed back Unicode
                  assert len(result.content) == 1
                  content = result.content[0]
                  assert content.type == "text"
                  assert f"Echo: {test_string}" == content.text, f"Failed for {test_name}"
  
  
  @pytest.mark.anyio
  async def test_streamable_http_client_unicode_prompts(running_unicode_server: str) -> None:
      """Test that Unicode text is correctly handled in prompts via streamable HTTP."""
      base_url = running_unicode_server
      endpoint_url = f"{base_url}/mcp"
  
      async with streamable_http_client(endpoint_url) as (read_stream, write_stream, _get_session_id):
          async with ClientSession(read_stream, write_stream) as session:
              await session.initialize()
  
              # Test 1: List prompts (server→client Unicode in descriptions)
              prompts = await session.list_prompts()
              assert len(prompts.prompts) == 1
  
              prompt = prompts.prompts[0]
              assert prompt.name == "unicode_prompt"
              assert prompt.description is not None
              assert "Слой хранилища, где располагаются" in prompt.description
  
              # Test 2: Get prompt with Unicode content (server→client)
              result = await session.get_prompt("unicode_prompt", arguments={})
              assert len(result.messages) == 1
  
              message = result.messages[0]
              assert message.role == "user"
              assert message.content.type == "text"
              assert message.content.text == "Hello世界🌍Привет안녕مرحباשלום"
  