Function unicode_session

tests/client/test_http_unicode.py:99–124

Yield an initialized ClientSession speaking streamable HTTP (SSE responses) to the Unicode test server, entirely in process.

()

@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,
    )
    # 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