feat!: adopt MCP revision 2026-07-28 via the mcp 2.x SDK

CHANGELOG.md

@@ -7,6 +7,36 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- **Adopt MCP revision `2026-07-28` (the stateless revision) via the `mcp` 2.x
+  SDK** (`mcp>=1.28.1,<2.0.0` → `mcp>=2.0.0,<3.0`). `FastMCP` → `MCPServer`;
+  `Context` now imports from `mcp.server.mcpserver`. Every tool's behaviour is
+  unchanged and a v2 server still serves 2025-era clients from the same process.
+  Transport constructor args (`host`/`port`) moved to the run call, matching the
+  SDK. Verified against `mcp-conformance` 0.2.0 (17 contracts, 1 injection skip).
+- **Offloaded-result and stored-response access is now authorized by possession
+  of the handle, not by session identity.** The 2026-07-28 revision removed
+  protocol sessions, so the previous `id(ctx.session)` scoping no longer works
+  (a fresh session object exists per request). Per the spec, cross-call state is
+  addressed by server-minted handles passed as tool arguments: the `retrieve_key`
+  is now an unguessable random capability token (was a content hash), and holding
+  a `retrieve_key` / `response_id` is the authorization to fetch it.
+
+### Fixed
+
+- **The stdio transport crashed on any JSON-RPC request larger than 64 KiB.** The
+  stdin reader used a default-limit `asyncio.StreamReader`; an over-64K line
+  raised `LimitOverrunError` and tore down the whole transport task group, taking
+  the server down on a single large request (legitimate or hostile). The reader
+  now uses a 16 MiB limit and turns a still-oversized line into a clean protocol
+  error instead of an unhandled crash. Regression-tested over a real subprocess.
+- Custom stdio transport now parses lines through the SDK's
+  `jsonrpc_message_adapter` (in 2.x `JSONRPCMessage` is a plain union alias with
+  no `model_validate_json`) and serializes replies with `exclude_unset=True`,
+  which keeps the 2026-07-28 envelope fields (`resultType`, `ttlMs`,
+  `cacheScope`) on the wire.
+
 ## [0.3.1] - 2026-07-28
 
 ### Security

conformance.toml

@@ -2,13 +2,20 @@
 # Run from the repo root with the project venv present:
 #   /path/to/mcp-conformance/.venv/bin/mcp-conformance run
 #
-# All three tools (perplexity_search, sonar_ask, deep_research) call the paid
-# Perplexity API, so there are NO behavioural probes — invoking them would cost
-# money and need a real key. The protocol/schema contracts still run in full;
-# the behavioural contracts skip by design. A dummy key lets the server start
+# The research tools (perplexity_search, sonar_ask, deep_research, responses_*)
+# call the paid Perplexity API, so they are NOT probed — invoking them would cost
+# money and need a real key. The behavioural probe is restricted to
+# server_metrics: a read-only tool that reports this process's request/latency
+# counters and never touches the API, so the suite runs offline with a dummy key
 # (the key is a SecretStr validated only for presence at startup).
 command = [".venv/bin/perplexity-agent"]
 env = { PERPLEXITY_API_KEY = "test-dummy-key" }
 
+[[probe]]
+tool = "server_metrics"
+
 [expect]
 max_output_chars = 200000
+# PerplexityAgent tags fetched web content as untrusted rather than flagging
+# injection in its own tool inputs, and the flagging tools call the API, so
+# expect.injection is intentionally omitted.

pyproject.toml

@@ -11,7 +11,7 @@ keywords = ["mcp", "perplexity", "sonar", "research", "agent", "llm"]
 # Dependencies are pinned with compatible-release bounds; uv.lock fully locks them.
 # (NSA recommendation: choose supported projects, pin dependencies, track vulns.)
 dependencies = [
-    "mcp>=1.28.1,<2.0.0",
+    "mcp>=2.0.0,<3.0",
     "httpx>=0.28.1,<0.29.0",
     "pydantic>=2.13.0,<3.0.0",
     "pydantic-settings>=2.14.2,<3.0.0",

src/perplexity_agent/__main__.py

@@ -18,14 +18,21 @@
 
 import anyio
 import mcp.types as mcp_types
-from mcp.server.fastmcp import FastMCP
+from mcp.server.mcpserver import MCPServer
 from mcp.shared.message import SessionMessage
 
 from .config import Settings, load_settings
 from .server import build_server
 
+# Max bytes for a single JSON-RPC line on stdin. asyncio.StreamReader defaults to
+# a 64 KiB line limit and raises on anything larger, which would crash the whole
+# transport task group on an over-64K request — a legitimately large tool call, or
+# a hostile oversized one. Raise it to a generous-but-bounded cap so real traffic
+# passes and a runaway line still can't exhaust memory.
+_MAX_STDIN_LINE = 16 * 1024 * 1024  # 16 MiB
 
-async def _serve_stdio(mcp: FastMCP) -> None:
+
+async def _serve_stdio(mcp: MCPServer) -> None:
     """Serve stdio and close the output stream when the client input reaches EOF.
 
     MCP 1.28's stock ``stdio_server`` can wait forever on its transport pumps after
@@ -39,14 +46,29 @@ async def _serve_stdio(mcp: FastMCP) -> None:
 
     async def stdin_reader() -> None:
         loop = asyncio.get_running_loop()
-        reader = asyncio.StreamReader()
+        reader = asyncio.StreamReader(limit=_MAX_STDIN_LINE)
         protocol = asyncio.StreamReaderProtocol(reader)
         transport, _ = await loop.connect_read_pipe(lambda: protocol, sys.stdin.buffer)
         try:
             async with read_sender:
-                while line := await reader.readline():
+                while True:
+                    try:
+                        line = await reader.readline()
+                    except ValueError as exc:
+                        # A single line exceeded _MAX_STDIN_LINE. The overrun data
+                        # stays buffered and would re-raise on every retry, so we
+                        # surface one protocol error and stop reading — a clean end
+                        # to the transport, never an unhandled crash of the whole
+                        # task group (the previous behaviour). 16 MiB is far above
+                        # any legitimate JSON-RPC line, so this is the hostile case.
+                        await read_sender.send(exc)
+                        break
+                    if not line:
+                        break
                     try:
-                        message = mcp_types.JSONRPCMessage.model_validate_json(line)
+                        message = mcp_types.jsonrpc_message_adapter.validate_json(
+                            line, by_name=False
+                        )
                     except Exception as exc:
                         await read_sender.send(exc)
                         continue
@@ -59,18 +81,18 @@ async def stdout_writer() -> None:
         async with write_receiver:
             async for session_message in write_receiver:
                 rendered = session_message.message.model_dump_json(
-                    by_alias=True, exclude_none=True
+                    by_alias=True, exclude_unset=True
                 )
                 sys.stdout.write(rendered + "\n")
                 sys.stdout.flush()
                 await anyio.lowlevel.checkpoint()
 
     async def server_runner() -> None:
         try:
-            await mcp._mcp_server.run(
+            await mcp._lowlevel_server.run(
                 read_stream,
                 write_stream,
-                mcp._mcp_server.create_initialization_options(),
+                mcp._lowlevel_server.create_initialization_options(),
             )
         finally:
             finished.set()
@@ -84,11 +106,11 @@ async def server_runner() -> None:
         task_group.cancel_scope.cancel()
 
 
-def _run_stdio(mcp: FastMCP) -> None:
+def _run_stdio(mcp: MCPServer) -> None:
     anyio.run(_serve_stdio, mcp)
 
 
-def _run_http(mcp: FastMCP, settings: Settings) -> None:
+def _run_http(mcp: MCPServer, settings: Settings) -> None:
     """Run the streamable-HTTP transport behind mandatory bearer-token auth."""
     if not settings.http_auth_token:
         sys.exit(

src/perplexity_agent/efficiency.py

@@ -22,6 +22,7 @@
 
 import hashlib
 import json
+import secrets
 from collections import OrderedDict
 from collections.abc import Iterable, Mapping
 from typing import Any
@@ -159,45 +160,35 @@ class OffloadStore:
 
     def __init__(self, max_entries: int = 128) -> None:
         self.max_entries = max(1, max_entries)
-        self._store: OrderedDict[str, tuple[str, str | None]] = OrderedDict()
-
-    def stash(self, payload: str, *, owner: str | None = None) -> str:
-        """Store ``payload`` for ``owner`` and return its content hash.
-
-        ``owner`` scopes retrieval by salting the content-derived key. Identical
-        content in two sessions therefore occupies independent bounded entries,
-        and an owner set cannot grow without bound behind one popular payload.
+        self._store: OrderedDict[str, str] = OrderedDict()
+
+    def stash(self, payload: str) -> str:
+        """Store ``payload`` and return an unguessable retrieval key.
+
+        The key is a fresh random token, not a content hash, so it is a genuine
+        server-minted capability: possession of the key IS the authorization to
+        retrieve. MCP revision 2026-07-28 removed protocol sessions, so retrieval
+        can no longer be scoped to a session identity — the spec's replacement is
+        exactly this, an opaque handle passed back as an ordinary tool argument.
+        A random key also means an attacker cannot fabricate one by guessing the
+        content, which a content-hash key allowed.
         """
-        key_material = payload if owner is None else f"{owner}\0{payload}"
-        key = hashlib.sha256(key_material.encode("utf-8", "replace")).hexdigest()[:24]
-        if key in self._store:
-            stored_payload, stored_owner = self._store[key]
-            # A collision is fantastically unlikely at 96 bits, but never grant
-            # access to different content merely because its short hash matched.
-            if stored_payload != payload or stored_owner != owner:
-                raise ValueError("offload retrieval-key collision")
-            self._store.move_to_end(key)
-        else:
-            self._store[key] = (payload, owner)
-            while len(self._store) > self.max_entries:
-                self._store.popitem(last=False)  # evict oldest
+        key = secrets.token_hex(12)  # 96 bits, same width as the old short hash
+        self._store[key] = payload
+        while len(self._store) > self.max_entries:
+            self._store.popitem(last=False)  # evict oldest
         return key
 
-    def retrieve(self, key: str, *, owner: str | None = None) -> str | None:
-        """Return ``owner``'s payload, or ``None`` if absent, evicted, or foreign."""
-        entry = self._store.get(key)
-        if entry is None:
-            return None
-        payload, stored_owner = entry
-        return payload if owner == stored_owner else None
+    def retrieve(self, key: str) -> str | None:
+        """Return the payload for ``key``, or ``None`` if absent or evicted."""
+        return self._store.get(key)
 
 
 def bound_result(
     result: Any,
     *,
     max_chars: int = DEFAULT_MAX_CHARS,
     store: OffloadStore | None = None,
-    owner: str | None = None,
 ) -> Any:
     """Bound any tool result to ``max_chars``.
 
@@ -213,7 +204,7 @@ def bound_result(
             return {
                 "truncated": True,
                 "content": bounded,
-                "retrieve_key": store.stash(result, owner=owner),
+                "retrieve_key": store.stash(result),
             }
         return bounded
 
@@ -227,5 +218,5 @@ def bound_result(
         "content": bounded,
     }
     if store is not None:
-        envelope["retrieve_key"] = store.stash(serialized, owner=owner)
+        envelope["retrieve_key"] = store.stash(serialized)
     return envelope