Skip to content

Run SSE and Unicode transport tests in process instead of over sockets #2765

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
maxisbey merged 1 commit into from
Jun 2, 2026
+242 −455
Merged

Run SSE and Unicode transport tests in process instead of over sockets #2765

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
288 changes: 112 additions & 176 deletions tests/client/test_http_unicode.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,10 @@
(server→client and client→server) using the streamable HTTP transport.
"""

import multiprocessing
import socket
from collections.abc import AsyncGenerator, Generator
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

import httpx
import pytest
from starlette.applications import Starlette
from starlette.routing import Mount
Expand All @@ -19,7 +18,10 @@
from mcp.server import Server, ServerRequestContext
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from mcp.types import TextContent, Tool
from tests.test_helpers import wait_for_server
from tests.interaction.transports import StreamingASGITransport

# The in-process app is mounted at this origin purely so URLs are well-formed; nothing listens here.
BASE_URL = "http://127.0.0.1:8000"

# Test constants with various Unicode characters
UNICODE_TEST_STRINGS = {
Expand All @@ -41,197 +43,131 @@
}


def run_unicode_server(port: int) -> None: # pragma: no cover
"""Run the Unicode test server in a separate process."""
import uvicorn

# Need to recreate the server setup in this process
async def handle_list_tools(
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
) -> types.ListToolsResult:
return types.ListToolsResult(
tools=[
Tool(
name="echo_unicode",
description="🔤 Echo Unicode text - Hello 👋 World 🌍 - Testing 🧪 Unicode ✨",
input_schema={
"type": "object",
"properties": {
"text": {"type": "string", "description": "Text to echo back"},
},
"required": ["text"],
async def handle_list_tools(
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
) -> types.ListToolsResult:
return types.ListToolsResult(
tools=[
Tool(
name="echo_unicode",
description="🔤 Echo Unicode text - Hello 👋 World 🌍 - Testing 🧪 Unicode ✨",
input_schema={
"type": "object",
"properties": {
"text": {"type": "string", "description": "Text to echo back"},
},
),
]
)

async def handle_call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> types.CallToolResult:
if params.name == "echo_unicode":
text = params.arguments.get("text", "") if params.arguments else ""
return types.CallToolResult(
content=[
TextContent(
type="text",
text=f"Echo: {text}",
)
]
"required": ["text"],
},
),
]
)


async def handle_call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> types.CallToolResult:
assert params.name == "echo_unicode"
assert params.arguments is not None
return types.CallToolResult(content=[TextContent(type="text", text=f"Echo: {params.arguments['text']}")])


async def handle_list_prompts(
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
) -> types.ListPromptsResult:
return types.ListPromptsResult(
prompts=[
types.Prompt(
name="unicode_prompt",
description="Unicode prompt - Слой хранилища, где располагаются",
arguments=[],
)
else:
raise ValueError(f"Unknown tool: {params.name}")

async def handle_list_prompts(
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
) -> types.ListPromptsResult:
return types.ListPromptsResult(
prompts=[
types.Prompt(
name="unicode_prompt",
description="Unicode prompt - Слой хранилища, где располагаются",
arguments=[],
)
]
)

async def handle_get_prompt(
ctx: ServerRequestContext, params: types.GetPromptRequestParams
) -> types.GetPromptResult:
if params.name == "unicode_prompt":
return types.GetPromptResult(
messages=[
types.PromptMessage(
role="user",
content=types.TextContent(
type="text",
text="Hello世界🌍Привет안녕مرحباשלום",
),
)
]
]
)


async def handle_get_prompt(ctx: ServerRequestContext, params: types.GetPromptRequestParams) -> types.GetPromptResult:
assert params.name == "unicode_prompt"
return types.GetPromptResult(
messages=[
types.PromptMessage(
role="user",
content=types.TextContent(type="text", text="Hello世界🌍Привет안녕مرحباשלום"),
)
raise ValueError(f"Unknown prompt: {params.name}")
]
)


@asynccontextmanager
async def unicode_session() -> AsyncIterator[ClientSession]:
"""Yield an initialized ClientSession speaking streamable HTTP (SSE responses) to the
Unicode test server, entirely in process."""
server = Server(
name="unicode_test_server",
on_list_tools=handle_list_tools,
on_call_tool=handle_call_tool,
on_list_prompts=handle_list_prompts,
on_get_prompt=handle_get_prompt,
)

# 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)
# SSE response mode, so Unicode rides the SSE event encoding rather than a plain JSON body.
session_manager = StreamableHTTPSessionManager(app=server, json_response=False)
app = Starlette(routes=[Mount("/mcp", app=session_manager.handle_request)])

async with (
session_manager.run(),
# follow_redirects matches the SDK's own client factory; Starlette's Mount 307-redirects
# the bare /mcp path to /mcp/.
httpx.AsyncClient(
transport=StreamingASGITransport(app), base_url=BASE_URL, follow_redirects=True
) as http_client,
streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client) as (read_stream, write_stream),
ClientSession(read_stream, write_stream) as session,
):
await session.initialize()
yield session


@pytest.mark.anyio
async def test_streamable_http_client_unicode_tool_call(running_unicode_server: str) -> None:
async def test_streamable_http_client_unicode_tool_call() -> 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):
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
async with unicode_session() as session:
# 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
# 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})
# 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}"
# 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:
async def test_streamable_http_client_unicode_prompts() -> 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):
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世界🌍Привет안녕مرحباשלום"
async with unicode_session() as session:
# 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世界🌍Привет안녕مرحباשלום"
Loading
Loading