"""Transcript tests for the minimal MCP JSON-RPC server."""
from __future__ import annotations
import asyncio
import io
import json
import unittest
from src.mcp_server import MAX_LINE_BYTES, McpServer
class FakeTools:
"""Provide deterministic tool results for protocol tests."""
async def call(self, name, arguments, cancelled):
"""Return the received call without browser access."""
if name == "cancellable":
while not cancelled():
await asyncio.sleep(0)
return {
"content": [{"type": "text", "text": "cancelled"}],
"structuredContent": {"cancelled": True},
"isError": True,
}
if name == "slow":
await asyncio.sleep(0.01)
return {
"content": [{"type": "text", "text": name}],
"structuredContent": arguments,
"isError": False,
}
def message(method, request_id=None, params=None, notification=False):
"""Encode one newline-delimited JSON-RPC message."""
value = {"jsonrpc": "2.0", "method": method}
if not notification:
value["id"] = request_id
if params is not None:
value["params"] = params
return json.dumps(value).encode() + b"\n"
class McpServerTest(unittest.IsolatedAsyncioTestCase):
"""Verify lifecycle, framing, errors, and stdout purity."""
async def transcript(self, data: bytes):
"""Run a full in-memory transcript and decode its responses."""
output = io.BytesIO()
server = McpServer(FakeTools(), output)
await server.run(io.BytesIO(data))
return [json.loads(line) for line in output.getvalue().splitlines()]
async def test_initializes_each_supported_version(self) -> None:
for version in (
"2025-11-25",
"2025-06-18",
"2024-11-05",
):
replies = await self.transcript(message(
"initialize",
1,
{"protocolVersion": version},
))
self.assertEqual(replies[0]["result"]["protocolVersion"], version)
async def test_negotiates_newest_for_unsupported_version(self) -> None:
replies = await self.transcript(message(
"initialize",
1,
{"protocolVersion": "1900-01-01"},
))
self.assertEqual(
replies[0]["result"]["protocolVersion"],
"2025-11-25",
)
async def test_notification_has_no_response_and_ping_works(self) -> None:
data = b"".join((
message("initialize", 1, {"protocolVersion": "2025-11-25"}),
message(
"notifications/initialized",
notification=True,
params={},
),
message("ping", 2, {}),
message("unknown/notification", notification=True),
))
replies = await self.transcript(data)
self.assertEqual({reply["id"] for reply in replies}, {1, 2})
ping = next(reply for reply in replies if reply["id"] == 2)
self.assertEqual(ping["result"], {})
async def test_lists_static_tools(self) -> None:
data = b"".join((
message("initialize", 1, {}),
message("tools/list", 2, {}),
))
replies = await self.transcript(data)
listing = next(reply for reply in replies if reply["id"] == 2)
tools = listing["result"]["tools"]
self.assertGreaterEqual(len(tools), 12)
self.assertIn(
"run_javascript",
{tool["name"] for tool in tools},
)
self.assertTrue(all(
tool["inputSchema"]["additionalProperties"] is False
for tool in tools
))
async def test_lists_tools_with_codex_request_metadata(self) -> None:
data = b"".join((
message(
"initialize",
1,
{"protocolVersion": "2025-06-18"},
),
message(
"tools/list",
2,
{"_meta": {"progressToken": 0}},
),
message(
"tools/list",
3,
{
"cursor": "next-page",
"_meta": {"progressToken": 1},
},
),
))
replies = await self.transcript(data)
initialization = next(reply for reply in replies if reply["id"] == 1)
self.assertEqual(
initialization["result"]["protocolVersion"],
"2025-06-18",
)
listing = next(reply for reply in replies if reply["id"] == 2)
self.assertIn("tools", listing["result"])
paginated = next(reply for reply in replies if reply["id"] == 3)
self.assertIn("tools", paginated["result"])
async def test_rejects_invalid_tools_list_params(self) -> None:
data = b"".join((
message("initialize", 1, {}),
message("tools/list", 2, {"unexpected": True}),
message("tools/list", 3, {"cursor": 1}),
message("tools/list", 4, {"_meta": "invalid"}),
))
replies = await self.transcript(data)
for request_id in (2, 3, 4):
reply = next(
item for item in replies if item["id"] == request_id
)
self.assertEqual(reply["error"]["code"], -32602)
async def test_tool_call_has_structured_content(self) -> None:
data = b"".join((
message("initialize", 1, {}),
message(
"tools/call",
2,
{
"name": "example",
"arguments": {"x": 1},
"_meta": {"progressToken": 0},
},
),
))
replies = await self.transcript(data)
call = next(reply for reply in replies if reply["id"] == 2)
self.assertEqual(call["result"]["structuredContent"], {"x": 1})
async def test_rejects_invalid_tool_call_meta(self) -> None:
data = b"".join((
message("initialize", 1, {}),
message(
"tools/call",
2,
{
"name": "example",
"arguments": {},
"_meta": "invalid",
},
),
))
replies = await self.transcript(data)
call = next(reply for reply in replies if reply["id"] == 2)
self.assertEqual(call["error"]["code"], -32602)
async def test_json_rpc_errors(self) -> None:
data = b"".join((
b"not json\n",
b"[]\n",
json.dumps({"jsonrpc": "1.0", "id": 3,
"method": "ping"}).encode() + b"\n",
message("initialize", 4, {}),
message("missing", 5, {}),
message("ping", 6, []),
))
replies = await self.transcript(data)
codes = [reply["error"]["code"] for reply in replies
if "error" in reply]
self.assertIn(-32700, codes)
self.assertIn(-32600, codes)
self.assertIn(-32601, codes)
self.assertIn(-32602, codes)
async def test_concurrent_responses_are_whole_lines(self) -> None:
calls = [message("initialize", 0, {})]
calls.extend(
message("tools/call", index, {
"name": "slow" if index % 2 else "fast",
"arguments": {"index": index},
})
for index in range(1, 20)
)
replies = await self.transcript(b"".join(calls))
self.assertEqual(len(replies), 20)
self.assertEqual({reply["id"] for reply in replies}, set(range(20)))
async def test_cancel_notification_reaches_active_tool(self) -> None:
data = b"".join((
message("initialize", 1, {}),
message(
"tools/call",
2,
{"name": "cancellable", "arguments": {}},
),
message(
"notifications/cancelled",
notification=True,
params={"requestId": 2},
),
))
replies = await self.transcript(data)
call = next(reply for reply in replies if reply["id"] == 2)
self.assertTrue(call["result"]["structuredContent"]["cancelled"])
async def test_oversized_line_is_rejected(self) -> None:
replies = await self.transcript(b" " * (MAX_LINE_BYTES + 1) + b"\n")
self.assertEqual(replies[0]["error"]["code"], -32700)
if __name__ == "__main__":
unittest.main()