MCP tests: the SDK's own client connects over streamable HTTP with a bearer, and older protocol revisions still answer

backend/tests/mcp/test_mcp_http.py

@@ -10,6 +10,8 @@ from typing import Any
 import httpx
 import pytest
 from fastapi.testclient import TestClient
+from mcp import ClientSession
+from mcp.client.streamable_http import streamable_http_client
 from sqlmodel import Session
 
 from fluksio.api.deps import CODE_IS_SUPERUSERS
@@ -65,6 +67,8 @@ def over_mcp(monkeypatch: pytest.MonkeyPatch):
 
         return asyncio.run(main())
 
+    # For a test that opens connections of its own, as a real client does.
+    run.app = mcp_app  # type: ignore[attr-defined]
     return run
 
 
@@ -249,3 +253,76 @@ def test_an_agent_cannot_write_code_its_person_may_not(
 
     answer = json.loads(over_mcp(block).json()["result"]["content"][0]["text"])
     assert answer == {"error": CODE_IS_SUPERUSERS, "status": 403}
+
+
+def test_a_real_client_can_connect_and_call_a_tool(
+    over_mcp,
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    agent: uuid.UUID,
+) -> None:
+    """The SDK's own client, over the transport an agent uses.
+
+    It does what the raw calls above leave out: initialize, the initialized
+    notification and the standalone-stream probe, with the bearer on every hop.
+    """
+    me = client.get(
+        f"{settings.API_V1_STR}/users/me", headers=superuser_token_headers
+    ).json()
+    token = mcp_token(uuid.UUID(me["id"]), agent)
+
+    async def block(_http: httpx.AsyncClient) -> tuple[Any, Any, Any]:
+        async with httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=over_mcp.app),
+            headers={"Authorization": f"Bearer {token}"},
+        ) as http_client:
+            async with streamable_http_client(
+                f"{settings.oauth_issuer}/mcp", http_client=http_client
+            ) as (read, write, _):
+                async with ClientSession(read, write) as session:
+                    initialized = await session.initialize()
+                    tools = await session.list_tools()
+                    called = await session.call_tool("list_flows", {})
+        return initialized, tools, called
+
+    initialized, tools, called = over_mcp(block)
+    assert initialized.serverInfo.name == "fluksio"
+    assert {"list_flows", "run_flow"} <= {tool.name for tool in tools.tools}
+    assert called.isError is False
+
+
+@pytest.mark.parametrize("revision", ["2025-03-26", "2025-06-18", "2025-11-25"])
+def test_clients_of_older_protocol_revisions_still_connect(
+    over_mcp,
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    agent: uuid.UUID,
+    revision: str,
+) -> None:
+    """Agents in the field speak whatever revision they shipped with."""
+    me = client.get(
+        f"{settings.API_V1_STR}/users/me", headers=superuser_token_headers
+    ).json()
+    token = mcp_token(uuid.UUID(me["id"]), agent)
+
+    async def block(http: httpx.AsyncClient) -> tuple[Any, Any]:
+        opening = rpc(
+            token,
+            "initialize",
+            protocolVersion=revision,
+            capabilities={},
+            clientInfo={"name": "test", "version": "1"},
+        )
+        initialized = await http.post("/mcp", **opening)
+        later = {**opening["headers"], "MCP-Protocol-Version": revision}
+        listed = await http.post(
+            "/mcp",
+            json={"jsonrpc": "2.0", "id": 2, "method": "tools/list"},
+            headers=later,
+        )
+        return initialized, listed
+
+    initialized, listed = over_mcp(block)
+    assert initialized.json()["result"]["protocolVersion"] == revision
+    assert listed.status_code == 200
+    assert "list_flows" in {tool["name"] for tool in listed.json()["result"]["tools"]}