Test case showing failure at main

Author: aabmass

tests/server/test_streamable_http_streams_closed.py

@@ -0,0 +1,50 @@
+from typing import Any
+
+import httpx
+import pytest
+
+import mcp.types as types
+from mcp.client.session import ClientSession
+from mcp.client.streamable_http import streamable_http_client
+from mcp.server import Server
+
+
+@pytest.fixture
+def server() -> Server:
+    server_app = Server("test_server")
+
+    @server_app.call_tool()
+    async def my_tool(name: str, arguments: dict[str, Any]) -> list[types.TextContent]:
+        return [types.TextContent(type="text", text="Tool result")]
+
+    @server_app.list_tools()
+    async def list_tools():
+        return [
+            types.Tool(
+                name="my_tool",
+                description="Test tool",
+                input_schema={
+                    "type": "object",
+                    "properties": {},
+                },
+            )
+        ]
+
+    return server_app
+
+
+HOST = "testserver"
+
+
+@pytest.mark.anyio
+async def test_streamable_http_server_cleanup(server: Server):
+    mcp_app = server.streamable_http_app(host=HOST)
+    async with (
+        mcp_app.router.lifespan_context(mcp_app),
+        httpx.ASGITransport(mcp_app) as transport,
+        httpx.AsyncClient(transport=transport) as client,
+        streamable_http_client(f"http://{HOST}/mcp", http_client=client) as (read_stream, write_stream),
+        ClientSession(read_stream, write_stream) as session,
+    ):
+        await session.initialize()
+        await session.call_tool("my_tool", arguments={})