From 088426934be2b610997d8de849aa3afb4d828bd4 Mon Sep 17 00:00:00 2001 From: Peter Alexander Date: Fri, 19 Sep 2025 17:59:46 +0100 Subject: [PATCH 1/2] Add comprehensive Unicode tests for streamable HTTP transport MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Test Unicode text transmission in both directions (client→server→client) - Verify Unicode handling in tool descriptions, arguments, and responses - Verify Unicode handling in prompt descriptions and content - Test with various Unicode scripts including Cyrillic, Chinese, Japanese, Korean, Arabic, Hebrew, Greek, emoji, and special characters - Use multiprocessing to run server in separate process (avoids ResourceWarnings) - Follow existing test patterns from test_streamable_http_manager.py --- tests/client/test_http_unicode.py | 247 ++++++++++++++++++++++++++++++ 1 file changed, 247 insertions(+) create mode 100644 tests/client/test_http_unicode.py diff --git a/tests/client/test_http_unicode.py b/tests/client/test_http_unicode.py new file mode 100644 index 000000000..5d243dedd --- /dev/null +++ b/tests/client/test_http_unicode.py @@ -0,0 +1,247 @@ +""" +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. +""" + +from collections.abc import Generator + +import pytest + +from mcp.client.session import ClientSession +from mcp.client.streamable_http import streamablehttp_client + +# 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: + """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.""" + import socket + + 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.""" + import multiprocessing + import socket + import time + + proc = multiprocessing.Process(target=run_unicode_server, kwargs={"port": unicode_server_port}, daemon=True) + proc.start() + + # Wait for server to be running + max_attempts = 20 + attempt = 0 + while attempt < max_attempts: + try: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.connect(("127.0.0.1", unicode_server_port)) + break + except ConnectionRefusedError: + time.sleep(0.1) + attempt += 1 + else: + raise RuntimeError(f"Server failed to start after {max_attempts} attempts") + + 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(): + 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 streamablehttp_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 streamablehttp_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世界🌍Привет안녕مرحباשלום" From fad7e0175c8d290ef80e3a45681fdefb88767661 Mon Sep 17 00:00:00 2001 From: Peter Alexander Date: Mon, 29 Sep 2025 13:33:53 +0100 Subject: [PATCH 2/2] Move imports to top of file in Unicode test Consolidate multiprocessing, socket, and time imports at the top of the file rather than importing them locally within fixtures. The imports within run_unicode_server remain as they must be available in the separate process context. --- tests/client/test_http_unicode.py | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/tests/client/test_http_unicode.py b/tests/client/test_http_unicode.py index 5d243dedd..edf8675e5 100644 --- a/tests/client/test_http_unicode.py +++ b/tests/client/test_http_unicode.py @@ -5,6 +5,9 @@ (server→client and client→server) using the streamable HTTP transport. """ +import multiprocessing +import socket +import time from collections.abc import Generator import pytest @@ -144,8 +147,6 @@ async def lifespan(app: Starlette) -> AsyncGenerator[None, None]: @pytest.fixture def unicode_server_port() -> int: """Find an available port for the Unicode test server.""" - import socket - with socket.socket() as s: s.bind(("127.0.0.1", 0)) return s.getsockname()[1] @@ -154,10 +155,6 @@ def unicode_server_port() -> int: @pytest.fixture def running_unicode_server(unicode_server_port: int) -> Generator[str, None, None]: """Start a Unicode test server in a separate process.""" - import multiprocessing - import socket - import time - proc = multiprocessing.Process(target=run_unicode_server, kwargs={"port": unicode_server_port}, daemon=True) proc.start()