Merge pull request chigwell#276 from omid-io/feat/mcp-protocol-2026-07-28

feat(protocol): support MCP 2026-07-28 and dual transport negotiation (chigwell#262)

Author: chigwell

pyproject.toml

@@ -25,7 +25,7 @@ requires-python = ">=3.10"
 dependencies = [
     "dotenv>=0.9.9",
     "httpx>=0.28.1",
-    "mcp[cli]>=1.8.0,<2",
+    "mcp[cli]>=1.30.0,<2",
     "pillow>=11.0.0",
     "pypdf>=4.0.0",
     "python-dotenv>=1.1.0",
@@ -54,7 +54,7 @@ whisper = [
 
 [tool.setuptools]
 py-modules = ["main", "migrate_session", "sanitize", "session_string_generator"]
-packages = ["telegram_mcp", "telegram_mcp.tools"]
+packages = ["telegram_mcp", "telegram_mcp.tools", "telegram_mcp.transport"]
 
 [project.scripts]
 telegram-mcp = "main:main"

requirements.txt

@@ -1,6 +1,6 @@
 dotenv>=0.9.9
 httpx>=0.28.1
-mcp[cli]>=1.8.0,<2
+mcp[cli]>=1.30.0,<2
 pillow>=11.0.0
 python-dotenv>=1.1.0
 python-json-logger>=3.3.0

telegram_mcp/runtime.py

@@ -153,6 +153,11 @@ def parse_schedule_date(
 TELEGRAM_API_ID = int(os.getenv("TELEGRAM_API_ID"))
 TELEGRAM_API_HASH = os.getenv("TELEGRAM_API_HASH")
 
+# Install protocol compatibility for MCP 2026-07-28 and backwards compatibility
+from telegram_mcp.transport import install_protocol_compatibility
+
+install_protocol_compatibility()
+
 # The shared HTTP service can be consumed by long-lived MCP clients. Stateless requests keep
 # those clients usable across server-process restarts instead of rejecting their next call
 # with "No valid session ID provided". Stdio transport remains unaffected.

telegram_mcp/server.py

@@ -0,0 +1,62 @@
+"""High-level server module and exports for Telegram MCP.
+
+Provides access to the FastMCP instance, HTTP ASGI application wrapped with
+protocol negotiation for MCP 2026-07-28, and transport serving routines.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from telegram_mcp.transport import (
+    DEFAULT_PROTOCOL_VERSION,
+    LATEST_PROTOCOL_VERSION,
+    SUPPORTED_PROTOCOL_VERSIONS,
+    create_streamable_http_app,
+    install_protocol_compatibility,
+    run_stdio_async,
+    run_streamable_http_async,
+)
+from telegram_mcp.runtime import mcp
+from telegram_mcp import runner as _runner
+
+# Ensure protocol compatibility is installed upon server module import
+install_protocol_compatibility()
+
+__all__ = [
+    "DEFAULT_PROTOCOL_VERSION",
+    "LATEST_PROTOCOL_VERSION",
+    "SUPPORTED_PROTOCOL_VERSIONS",
+    "create_streamable_http_app",
+    "get_http_app",
+    "get_server",
+    "main",
+    "mcp",
+    "run_stdio_async",
+    "run_streamable_http_async",
+    "serve",
+]
+
+
+def get_server() -> Any:
+    """Return the underlying FastMCP server instance."""
+    return mcp
+
+
+def get_http_app() -> Any:
+    """Return the Starlette ASGI application with protocol negotiation."""
+    return create_streamable_http_app(mcp)
+
+
+async def serve(transport: str = "stdio") -> None:
+    """Serve the Telegram MCP server over the requested transport."""
+    await _runner._serve(transport)
+
+
+def main() -> None:
+    """Server CLI entrypoint."""
+    _runner.main()
+
+
+if __name__ == "__main__":
+    main()

telegram_mcp/transport/__init__.py

@@ -0,0 +1,42 @@
+"""Transport layer and MCP protocol negotiation for telegram-mcp."""
+
+from telegram_mcp.transport.protocol import (
+    DEFAULT_PROTOCOL_VERSION,
+    HANDSHAKE_PROTOCOL_VERSIONS,
+    LATEST_PROTOCOL_VERSION,
+    MCP_PROTOCOL_VERSION_HEADER,
+    MODERN_PROTOCOL_VERSIONS,
+    OLDEST_SUPPORTED_VERSION,
+    SUPPORTED_PROTOCOL_VERSIONS,
+    UnsupportedProtocolVersionError,
+    install_protocol_compatibility,
+    is_supported_protocol_version,
+    negotiate_protocol_version,
+)
+from telegram_mcp.transport.http import (
+    ProtocolNegotiationMiddleware,
+    create_streamable_http_app,
+    run_streamable_http_async,
+)
+from telegram_mcp.transport.stdio import run_stdio_async
+
+# Ensure runtime patches are applied upon transport import
+install_protocol_compatibility()
+
+__all__ = [
+    "DEFAULT_PROTOCOL_VERSION",
+    "HANDSHAKE_PROTOCOL_VERSIONS",
+    "LATEST_PROTOCOL_VERSION",
+    "MCP_PROTOCOL_VERSION_HEADER",
+    "MODERN_PROTOCOL_VERSIONS",
+    "OLDEST_SUPPORTED_VERSION",
+    "SUPPORTED_PROTOCOL_VERSIONS",
+    "UnsupportedProtocolVersionError",
+    "install_protocol_compatibility",
+    "is_supported_protocol_version",
+    "negotiate_protocol_version",
+    "ProtocolNegotiationMiddleware",
+    "create_streamable_http_app",
+    "run_streamable_http_async",
+    "run_stdio_async",
+]

telegram_mcp/transport/http.py

@@ -0,0 +1,199 @@
+"""Streamable HTTP transport with MCP 2026-07-28 protocol negotiation support."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from starlette.responses import JSONResponse, Response
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
+
+from telegram_mcp.transport.protocol import (
+    DEFAULT_PROTOCOL_VERSION,
+    LATEST_PROTOCOL_VERSION,
+    MCP_PROTOCOL_VERSION_HEADER,
+    SUPPORTED_PROTOCOL_VERSIONS,
+    install_protocol_compatibility,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class ProtocolNegotiationMiddleware:
+    """ASGI middleware that enforces MCP protocol negotiation and response headers.
+
+    Features:
+    - Validates MCP-Protocol-Version request header against supported versions.
+    - Rejects unsupported versions with a standard JSON-RPC 400 Bad Request error.
+    - Injects negotiated MCP-Protocol-Version into every HTTP response.
+    - Handles CORS and OPTIONS preflight requests gracefully.
+    - Provides a JSON discovery response for plain GET probes (non-SSE) on the MCP endpoint.
+    """
+
+    def __init__(self, app: ASGIApp):
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        headers = dict(scope.get("headers", []))
+        raw_version_bytes = headers.get(MCP_PROTOCOL_VERSION_HEADER.encode("latin-1"))
+        requested_version = (
+            raw_version_bytes.decode("latin-1").strip() if raw_version_bytes else None
+        )
+
+        # 1. Handle OPTIONS discovery / CORS preflight
+        if scope.get("method") == "OPTIONS":
+            cors_headers = (
+                "Content-Type, Accept, MCP-Protocol-Version, MCP-Session-Id, Authorization"
+            )
+            response = Response(
+                status_code=204,
+                headers={
+                    "Allow": "GET, POST, DELETE, OPTIONS, HEAD",
+                    MCP_PROTOCOL_VERSION_HEADER: requested_version or LATEST_PROTOCOL_VERSION,
+                    "Access-Control-Allow-Origin": "*",
+                    "Access-Control-Allow-Methods": "GET, POST, DELETE, OPTIONS, HEAD",
+                    "Access-Control-Allow-Headers": cors_headers,
+                    "Access-Control-Expose-Headers": "MCP-Protocol-Version, MCP-Session-Id",
+                },
+            )
+            await response(scope, receive, send)
+            return
+
+        # 2. Validate protocol version if provided
+        if requested_version:
+            if requested_version not in SUPPORTED_PROTOCOL_VERSIONS:
+                supported_str = ", ".join(SUPPORTED_PROTOCOL_VERSIONS)
+                error_body = {
+                    "jsonrpc": "2.0",
+                    "id": "server-error",
+                    "error": {
+                        "code": -32600,
+                        "message": (
+                            f"Bad Request: Unsupported protocol version: {requested_version}. "
+                            f"Supported versions: {supported_str}"
+                        ),
+                    },
+                }
+                response = JSONResponse(
+                    status_code=400,
+                    content=error_body,
+                    headers={
+                        MCP_PROTOCOL_VERSION_HEADER: DEFAULT_PROTOCOL_VERSION,
+                    },
+                )
+                await response(scope, receive, send)
+                return
+            negotiated_version = requested_version
+        else:
+            negotiated_version = DEFAULT_PROTOCOL_VERSION
+
+        # 3. Handle plain GET discovery probe (clients querying without text/event-stream)
+        accept_header = headers.get(b"accept", b"").decode("latin-1").lower()
+        if scope.get("method") == "GET" and "text/event-stream" not in accept_header:
+            discovery_data = {
+                "name": "telegram",
+                "version": "2.0.1",
+                "protocolVersions": list(SUPPORTED_PROTOCOL_VERSIONS),
+                "latestProtocolVersion": LATEST_PROTOCOL_VERSION,
+                "negotiatedProtocolVersion": negotiated_version,
+            }
+            response = JSONResponse(
+                status_code=200,
+                content=discovery_data,
+                headers={
+                    MCP_PROTOCOL_VERSION_HEADER: negotiated_version,
+                    "Access-Control-Allow-Origin": "*",
+                    "Access-Control-Expose-Headers": "MCP-Protocol-Version, MCP-Session-Id",
+                },
+            )
+            await response(scope, receive, send)
+            return
+
+        # 4. Wrap send to inject MCP-Protocol-Version header on all response start messages
+        async def send_with_protocol_header(message: Message) -> None:
+            if message["type"] == "http.response.start":
+                resp_headers = list(message.get("headers", []))
+                header_names = {h[0].lower() for h in resp_headers}
+
+                if MCP_PROTOCOL_VERSION_HEADER.encode("latin-1") not in header_names:
+                    resp_headers.append(
+                        (
+                            MCP_PROTOCOL_VERSION_HEADER.encode("latin-1"),
+                            negotiated_version.encode("latin-1"),
+                        )
+                    )
+
+                if b"access-control-expose-headers" not in header_names:
+                    expose_val = b"*, mcp-protocol-version, mcp-session-id"
+                    resp_headers.append((b"access-control-expose-headers", expose_val))
+
+                message["headers"] = resp_headers
+            await send(message)
+
+        await self.app(scope, receive, send_with_protocol_header)
+
+
+def patch_fastmcp_streamable_http() -> None:
+    """Patch FastMCP.streamable_http_app to enforce MCP 2026-07-28 protocol negotiation."""
+    try:
+        from mcp.server.fastmcp import FastMCP
+
+        if getattr(FastMCP, "_protocol_negotiation_patched", False):
+            return
+
+        original_streamable_http_app = FastMCP.streamable_http_app
+
+        def wrapped_streamable_http_app(self: Any, *args: Any, **kwargs: Any) -> Any:
+            app = original_streamable_http_app(self, *args, **kwargs)
+            return ProtocolNegotiationMiddleware(app)
+
+        FastMCP.streamable_http_app = wrapped_streamable_http_app  # type: ignore[assignment]
+        FastMCP._protocol_negotiation_patched = True  # type: ignore[attr-defined]
+    except (ImportError, AttributeError) as exc:
+        logger.debug("Could not monkey-patch FastMCP.streamable_http_app: %s", exc)
+
+
+def create_streamable_http_app(mcp_instance: Any) -> Any:
+    """Build and wrap the Starlette app from FastMCP with protocol negotiation."""
+    install_protocol_compatibility()
+    if not hasattr(mcp_instance, "streamable_http_app"):
+        return mcp_instance
+
+    # If the FastMCP session manager already ran and shut down, reset it to allow clean restart
+    session_mgr = getattr(mcp_instance, "_session_manager", None)
+    if session_mgr is not None and getattr(session_mgr, "_has_started", False):
+        mcp_instance._session_manager = None
+
+    app = mcp_instance.streamable_http_app()
+    # If not already wrapped via monkeypatch
+    if not isinstance(app, ProtocolNegotiationMiddleware):
+        app = ProtocolNegotiationMiddleware(app)
+    return app
+
+
+async def run_streamable_http_async(mcp_instance: Any) -> None:
+    """Run FastMCP streamable HTTP transport with protocol negotiation middleware."""
+    install_protocol_compatibility()
+    patch_fastmcp_streamable_http()
+
+    if not hasattr(mcp_instance, "streamable_http_app"):
+        if hasattr(mcp_instance, "run_streamable_http_async"):
+            await mcp_instance.run_streamable_http_async()
+            return
+
+    import uvicorn
+
+    app = create_streamable_http_app(mcp_instance)
+
+    config = uvicorn.Config(
+        app,
+        host=mcp_instance.settings.host,
+        port=mcp_instance.settings.port,
+        log_level=mcp_instance.settings.log_level.lower(),
+    )
+    server = uvicorn.Server(config)
+    await server.serve()