# -*- coding: utf-8 -*-

'''Location: ./mcpgateway/translate.py
Copyright 2026
SPDX-License-Identifier: Apache-2.0
Authors: Mihai Criveti, Manav Gupta

r"""Bridges between different MCP transport protocols.
This module provides bidirectional bridging between MCP servers that communicate
via different transport protocols: stdio/JSON-RPC, HTTP/SSE, and streamable HTTP.
It enables exposing local MCP servers over HTTP or consuming remote endpoints
as local stdio servers.

The bridge supports multiple modes of operation:
- stdio to SSE: Expose a local stdio MCP server over HTTP/SSE
- SSE to stdio: Bridge a remote SSE endpoint to local stdio
- stdio to streamable HTTP: Expose a local stdio MCP server via streamable HTTP
- streamable HTTP to stdio: Bridge a remote streamable HTTP endpoint to local stdio

Examples:
    Programmatic usage:

    >>> import asyncio
    >>> from mcpgateway.translate import start_stdio
    >>> asyncio.run(start_stdio("uvx mcp-server-git", 9000, "info", None, "127.0.0.1"))  # doctest: +SKIP

    Test imports and configuration:

    >>> from mcpgateway.translate import MCPServer, StreamableHTTPSessionManager
    >>> isinstance(MCPServer, type)
    True
    >>> isinstance(StreamableHTTPSessionManager, type)
    True
    >>> from mcpgateway.translate import KEEP_ALIVE_INTERVAL
    >>> KEEP_ALIVE_INTERVAL > 0
    True
    >>> from mcpgateway.translate import DEFAULT_KEEPALIVE_ENABLED
    >>> isinstance(DEFAULT_KEEPALIVE_ENABLED, bool)
    True

    Test Starlette imports:

    >>> from mcpgateway.translate import Starlette, Route
    >>> isinstance(Starlette, type)
    True
    >>> isinstance(Route, type)
    True

    Test logging setup:

    >>> from mcpgateway.translate import LOGGER, logging_service
    >>> LOGGER is not None
    True
    >>> logging_service is not None
    True
    >>> hasattr(LOGGER, 'info')
    True
    >>> hasattr(LOGGER, 'error')
    True
    >>> hasattr(LOGGER, 'debug')
    True

    Test utility classes:

    >>> from mcpgateway.translate import _PubSub, StdIOEndpoint
    >>> pubsub = _PubSub()
    >>> hasattr(pubsub, 'publish')
    True
    >>> hasattr(pubsub, 'subscribe')
    True
    >>> hasattr(pubsub, 'unsubscribe')
    True

Usage:
    Command line usage::

        # 1. Expose an MCP server that talks JSON-RPC on stdio at :9000/sse
        python3 -m mcpgateway.translate --stdio "uvx mcp-server-git" --port 9000

        # 2. Bridge a remote SSE endpoint to local stdio
        python3 -m mcpgateway.translate --sse "https://example.com/sse" \
                --stdioCommand "uvx mcp-client"

        # 3. Expose stdio server via streamable HTTP at :9000/mcp
        python3 -m mcpgateway.translate --streamableHttp "uvx mcp-server-git" \
                --port 9000 --stateless --jsonResponse

        # 4. Connect to remote streamable HTTP endpoint
        python3 -m mcpgateway.translate \
                --streamableHttp "https://example.com/mcp" \
                --oauth2Bearer "your-token"

        # 5. Test SSE endpoint
        curl -N http://localhost:9000/sse          # receive the stream

        # 6. Send a test echo request to SSE endpoint
        curl -X POST http://localhost:9000/message \
             -H 'Content-Type: application/json'   \
             -d '{"jsonrpc":"2.0","id":1,"method":"echo","params":{"value":"hi"}}'

        # 7. Test streamable HTTP endpoint
        curl -X POST http://localhost:9000/mcp \
             -H 'Content-Type: application/json' \
             -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"demo","version":"0.0.1"}}}'

    The SSE stream emits JSON-RPC responses as ``event: message`` frames and sends
    regular ``event: keepalive`` frames (default every 30s) to prevent timeouts.

    Streamable HTTP supports both stateful (with session management) and stateless
    modes, and can return either JSON responses or SSE streams.
"""
'''

# Future
from __future__ import annotations

# Standard
import argparse
import asyncio
from contextlib import suppress
import logging
import os
import shlex
import signal
import sys
from typing import Any, AsyncIterator, Dict, List, Optional, Sequence, Tuple
from urllib.parse import urlencode
import uuid

# Third-Party
import anyio
from fastapi import FastAPI, Request, Response, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import PlainTextResponse
from mcp.server import Server as MCPServer
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
import orjson
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.types import Receive, Scope, Send
import uvicorn

try:
    # Third-Party
    import httpx
except ImportError:
    httpx = None  # type: ignore[assignment]

# First-Party
from mcpgateway.services.logging_service import LoggingService
from mcpgateway.translate_header_utils import extract_env_vars_from_headers, NormalizedMappings, parse_header_mappings

# Use patched EventSourceResponse with CPU spin protection (anyio#695 fix)
from mcpgateway.transports.sse_transport import EventSourceResponse
from mcpgateway.utils.orjson_response import ORJSONResponse

# Initialize logging service first
logging_service = LoggingService()
LOGGER = logging_service.get_logger("mcpgateway.translate")
CONTENT_TYPE = os.getenv("FORGE_CONTENT_TYPE", "application/json")
# headers = {"Content-Type": CONTENT_TYPE}
# Import settings for default keepalive interval
try:
    # First-Party
    from mcpgateway.config import settings

    DEFAULT_KEEP_ALIVE_INTERVAL = settings.sse_keepalive_interval
    DEFAULT_KEEPALIVE_ENABLED = settings.sse_keepalive_enabled
    DEFAULT_SSL_VERIFY = not settings.skip_ssl_verify
except ImportError:
    # Fallback if config not available
    DEFAULT_KEEP_ALIVE_INTERVAL = 30
    DEFAULT_KEEPALIVE_ENABLED = True
    DEFAULT_SSL_VERIFY = True  # Verify SSL by default when config unavailable

KEEP_ALIVE_INTERVAL = DEFAULT_KEEP_ALIVE_INTERVAL  # seconds - from config or fallback to 30

# Buffer limit for subprocess stdout (default 64KB is too small for large tool responses)
# Set to 16MB to handle tools that return large amounts of data (e.g., search results)
STDIO_BUFFER_LIMIT = 16 * 1024 * 1024  # 16MB

__all__ = ["main"]  # for console-script entry-point


# ---------------------------------------------------------------------------#
# Helpers - trivial in-process Pub/Sub                                       #
# ---------------------------------------------------------------------------#
class _PubSub:
    """Very small fan-out helper - one async Queue per subscriber.

    This class implements a simple publish-subscribe pattern using asyncio queues
    for distributing messages from stdio subprocess to multiple SSE clients.

    Examples:
        >>> import asyncio
        >>> async def test_pubsub():
        ...     pubsub = _PubSub()
        ...     q = pubsub.subscribe()
        ...     await pubsub.publish("hello")
        ...     result = await q.get()
        ...     pubsub.unsubscribe(q)
        ...     return result
        >>> asyncio.run(test_pubsub())
        'hello'
    """

    def __init__(self) -> None:
        """Initialize a new publish-subscribe system.

        Creates an empty list of subscriber queues. Each subscriber will
        receive their own asyncio.Queue for receiving published messages.

        Examples:
            >>> pubsub = _PubSub()
            >>> isinstance(pubsub._subscribers, list)
            True
            >>> len(pubsub._subscribers)
            0
            >>> hasattr(pubsub, '_subscribers')
            True
        """
        self._subscribers: List[asyncio.Queue[str]] = []

    async def publish(self, data: str) -> None:
        """Publish data to all subscribers.

        Dead queues (full) are automatically removed from the subscriber list.

        Args:
            data: The data string to publish to all subscribers.

        Examples:
            >>> import asyncio
            >>> async def test_publish():
            ...     pubsub = _PubSub()
            ...     await pubsub.publish("test")  # No subscribers, no error
            ...     return True
            >>> asyncio.run(test_publish())
            True

            >>> # Test queue full handling
            >>> async def test_full_queue():
            ...     pubsub = _PubSub()
            ...     # Create a queue with size 1
            ...     q = asyncio.Queue(maxsize=1)
            ...     pubsub._subscribers = [q]
            ...     # Fill the queue
            ...     await q.put("first")
            ...     # This should remove the full queue
            ...     await pubsub.publish("second")
            ...     return len(pubsub._subscribers)
            >>> asyncio.run(test_full_queue())
            0
        """
        dead: List[asyncio.Queue[str]] = []
        for q in self._subscribers:
            try:
                q.put_nowait(data)
            except asyncio.QueueFull:
                dead.append(q)
        for q in dead:
            with suppress(ValueError):
                self._subscribers.remove(q)

    def subscribe(self) -> "asyncio.Queue[str]":
        """Subscribe to published data.

        Creates a new queue for receiving published messages with a maximum
        size of 1024 items.

        Returns:
            asyncio.Queue[str]: A queue that will receive published data.

        Examples:
            >>> pubsub = _PubSub()
            >>> q = pubsub.subscribe()
            >>> isinstance(q, asyncio.Queue)
            True
            >>> q.maxsize
            1024
            >>> len(pubsub._subscribers)
            1
            >>> pubsub._subscribers[0] is q
            True
        """
        q: asyncio.Queue[str] = asyncio.Queue(maxsize=1024)
        self._subscribers.append(q)
        return q

    def unsubscribe(self, q: "asyncio.Queue[str]") -> None:
        """Unsubscribe from published data.

        Removes the queue from the subscriber list. Safe to call even if
        the queue is not in the list.

        Args:
            q: The queue to unsubscribe from published data.

        Examples:
            >>> pubsub = _PubSub()
            >>> q = pubsub.subscribe()
            >>> pubsub.unsubscribe(q)
            >>> pubsub.unsubscribe(q)  # No error on double unsubscribe
        """
        with suppress(ValueError):
            self._subscribers.remove(q)


# ---------------------------------------------------------------------------#
# StdIO endpoint (child process ↔ async queues)                              #
# ---------------------------------------------------------------------------#
class StdIOEndpoint:
    """Wrap a child process whose stdin/stdout speak line-delimited JSON-RPC.

    This class manages a subprocess that communicates via stdio using JSON-RPC
    protocol, pumping messages between the subprocess and a pubsub system.

    Examples:
        >>> import asyncio
        >>> async def test_stdio():
        ...     pubsub = _PubSub()
        ...     stdio = StdIOEndpoint("echo hello", pubsub)
        ...     # Would start a real subprocess
        ...     return isinstance(stdio, StdIOEndpoint)
        >>> asyncio.run(test_stdio())
        True
    """

    def __init__(self, cmd: str, pubsub: _PubSub, env_vars: Optional[Dict[str, str]] = None, header_mappings: Optional[NormalizedMappings] = None) -> None:
        """Initialize a stdio endpoint for subprocess communication.

        Sets up the endpoint with the command to run and the pubsub system
        for message distribution. The subprocess is not started until start()
        is called.

        Args:
            cmd: The command string to execute as a subprocess.
            pubsub: The publish-subscribe system for distributing subprocess
                output to SSE clients.
            env_vars: Optional dictionary of environment variables to set
                when starting the subprocess.
            header_mappings: Optional mapping of HTTP headers to environment variable names
                for dynamic environment injection.

        Examples:
            >>> pubsub = _PubSub()
            >>> endpoint = StdIOEndpoint("echo hello", pubsub)
            >>> endpoint._cmd
            'echo hello'
            >>> endpoint._proc is None
            True
            >>> isinstance(endpoint._pubsub, _PubSub)
            True
            >>> endpoint._stdin is None
            True
            >>> endpoint._pump_task is None
            True
            >>> endpoint._pubsub is pubsub
            True
        """
        self._cmd = cmd
        self._pubsub = pubsub
        self._env_vars = env_vars or {}
        self._header_mappings = header_mappings
        self._proc: Optional[asyncio.subprocess.Process] = None
        self._stdin: Optional[asyncio.StreamWriter] = None
        self._pump_task: Optional[asyncio.Task[None]] = None

    async def start(self, additional_env_vars: Optional[Dict[str, str]] = None) -> None:
        """Start the stdio subprocess with custom environment variables.

        Creates the subprocess and starts the stdout pump task. The subprocess
        is created with stdin/stdout pipes and stderr passed through.

        Args:
            additional_env_vars: Optional dictionary of additional environment
                variables to set when starting the subprocess. These will be
                combined with the environment variables set during initialization.

        Raises:
            RuntimeError: If the subprocess fails to create stdin/stdout pipes.

        Examples:
            >>> import asyncio # doctest: +SKIP
            >>> async def test_start(): # doctest: +SKIP
            ...     pubsub = _PubSub()
            ...     stdio = StdIOEndpoint("cat", pubsub)
            ...     # await stdio.start()  # doctest: +SKIP
            ...     return True
            >>> asyncio.run(test_start()) # doctest: +SKIP
            True
        """
        # Stop existing subprocess before starting a new one
        if self._proc is not None:
            await self.stop()

        LOGGER.info(f"Starting stdio subprocess: {self._cmd}")

        # Build environment from base + configured + additional
        env = os.environ.copy()
        env.update(self._env_vars)
        if additional_env_vars:
            env.update(additional_env_vars)

        # System-critical environment variables that must never be cleared
        system_critical_vars = {"PATH", "HOME", "TMPDIR", "TEMP", "TMP", "USER", "LOGNAME", "SHELL", "LANG", "LC_ALL", "LC_CTYPE", "PYTHONHOME", "PYTHONPATH"}

        # Clear any mapped env vars that weren't provided in headers to avoid inheritance
        if self._header_mappings:
            for env_var_name in self._header_mappings.values():
                if env_var_name not in (additional_env_vars or {}) and env_var_name not in system_critical_vars:
                    # Delete the variable instead of setting to empty string to avoid
                    # breaking subprocess initialization
                    env.pop(env_var_name, None)

        LOGGER.debug(f"Subprocess environment variables: {list(env.keys())}")
        self._proc = await asyncio.create_subprocess_exec(
            *shlex.split(self._cmd),
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=sys.stderr,  # passthrough for visibility
            env=env,  # 🔑 Add environment variable support
            limit=STDIO_BUFFER_LIMIT,  # Increase buffer limit for large tool responses
        )

        # Explicit error checking
        if not self._proc.stdin or not self._proc.stdout:
            raise RuntimeError(f"Failed to create subprocess with stdin/stdout pipes for command: {self._cmd}")

        LOGGER.debug("Subprocess started successfully")

        self._stdin = self._proc.stdin
        self._pump_task = asyncio.create_task(self._pump_stdout())

    async def stop(self) -> None:
        """Stop the stdio subprocess.

        Terminates the subprocess gracefully with a 5-second timeout,
        then cancels the pump task.

        Examples:
            >>> import asyncio
            >>> async def test_stop():
            ...     pubsub = _PubSub()
            ...     stdio = StdIOEndpoint("cat", pubsub)
            ...     await stdio.stop()  # Safe to call even if not started
            ...     return True
            >>> asyncio.run(test_stop())
            True
        """
        if self._proc is None:
            return

        # Check if process is still running
        try:
            if self._proc.returncode is not None:
                # Process already terminated
                LOGGER.info(f"Subprocess (pid={self._proc.pid}) already terminated")
                self._proc = None
                self._stdin = None
                return
        except (ProcessLookupError, AttributeError):
            # Process doesn't exist or is already cleaned up
            LOGGER.info("Subprocess already cleaned up")
            self._proc = None
            self._stdin = None
            return

        LOGGER.info(f"Stopping subprocess (pid={self._proc.pid})")
        try:
            self._proc.terminate()
            with suppress(asyncio.TimeoutError):
                await asyncio.wait_for(self._proc.wait(), timeout=5)
        except ProcessLookupError:
            # Process already terminated
            LOGGER.info("Subprocess already terminated")
        finally:
            if self._pump_task:
                self._pump_task.cancel()
                # Wait for pump task to actually finish, suppressing all exceptions
                # since the pump task logs its own errors. Use BaseException to catch
                # CancelledError which inherits from BaseException in Python 3.8+
                with suppress(BaseException):
                    await self._pump_task
            self._proc = None
            self._stdin = None  # Reset stdin too!

    def is_running(self) -> bool:
        """Check if the stdio subprocess is currently running.

        Returns:
            True if the subprocess is running, False otherwise.

        Examples:
            >>> import asyncio
            >>> async def test_is_running():
            ...     pubsub = _PubSub()
            ...     stdio = StdIOEndpoint("cat", pubsub)
            ...     return stdio.is_running()
            >>> asyncio.run(test_is_running())
            False
        """
        return self._proc is not None

    async def send(self, raw: str) -> None:
        """Send data to the subprocess stdin.

        Args:
            raw: The raw data string to send to the subprocess.

        Raises:
            RuntimeError: If the stdio endpoint is not started.

        Examples:
            >>> import asyncio
            >>> async def test_send():
            ...     pubsub = _PubSub()
            ...     stdio = StdIOEndpoint("cat", pubsub)
            ...     try:
            ...         await stdio.send("test")
            ...     except RuntimeError as e:
            ...         return str(e)
            >>> asyncio.run(test_send())
            'stdio endpoint not started'
        """
        if not self._stdin:
            raise RuntimeError("stdio endpoint not started")
        LOGGER.debug(f"→ stdio: {raw.strip()}")
        self._stdin.write(raw.encode())
        await self._stdin.drain()

    async def _pump_stdout(self) -> None:
        """Pump stdout from subprocess to pubsub.

        Continuously reads lines from the subprocess stdout and publishes them
        to the pubsub system. Runs until EOF or exception.

        Raises:
            RuntimeError: If process or stdout is not properly initialized.
            Exception: For any other error encountered while pumping stdout.
        """
        if not self._proc or not self._proc.stdout:
            raise RuntimeError("Process not properly initialized: missing stdout")

        reader = self._proc.stdout
        try:
            while True:
                line = await reader.readline()
                if not line:  # EOF
                    break
                text = line.decode(errors="replace")
                LOGGER.debug(f"← stdio: {text.strip()}")
                await self._pubsub.publish(text)
        except ConnectionResetError:  # pragma: no cover --subprocess terminated
            # Subprocess terminated abruptly - this is expected behavior
            LOGGER.debug("stdout pump: subprocess connection closed")
        except Exception:  # pragma: no cover --best-effort logging
            LOGGER.exception("stdout pump crashed - terminating bridge")
            raise


# ---------------------------------------------------------------------------#
# SSE Event Parser                                                           #
# ---------------------------------------------------------------------------#
class SSEEvent:
    """Represents a Server-Sent Event with proper field parsing.

    Attributes:
        event: The event type (e.g., 'message', 'keepalive', 'endpoint')
        data: The event data payload
        event_id: Optional event ID
        retry: Optional retry interval in milliseconds
    """

    def __init__(self, event: str = "message", data: str = "", event_id: Optional[str] = None, retry: Optional[int] = None):
        """Initialize an SSE event.

        Args:
            event: Event type, defaults to "message"
            data: Event data payload
            event_id: Optional event ID
            retry: Optional retry interval in milliseconds
        """
        self.event = event
        self.data = data
        self.event_id = event_id
        self.retry = retry

    @classmethod
    def parse_sse_line(cls, line: str, current_event: Optional["SSEEvent"] = None) -> Tuple[Optional["SSEEvent"], bool]:
        """Parse a single SSE line and update or create an event.

        Args:
            line: The SSE line to parse
            current_event: The current event being built (if any)

        Returns:
            Tuple of (event, is_complete) where event is the SSEEvent object
            and is_complete indicates if the event is ready to be processed
        """
        line = line.rstrip("\n\r")

        # Empty line signals end of event
        if not line:
            if current_event and current_event.data:
                return current_event, True
            return None, False

        # Comment line
        if line.startswith(":"):
            return current_event, False

        # Parse field
        if ":" in line:
            field, value = line.split(":", 1)
            value = value.lstrip(" ")  # Remove leading space if present
        else:
            field = line
            value = ""

        # Create event if needed
        if current_event is None:
            current_event = cls()

        # Update fields
        if field == "event":
            current_event.event = value
        elif field == "data":
            if current_event.data:
                current_event.data += "\n" + value
            else:
                current_event.data = value
        elif field == "id":
            current_event.event_id = value
        elif field == "retry":
            try:
                current_event.retry = int(value)
            except ValueError:
                pass  # Ignore invalid retry values

        return current_event, False


# ---------------------------------------------------------------------------#
# FastAPI app exposing /sse  &  /message                                     #
# ---------------------------------------------------------------------------#


def _build_fastapi(
    pubsub: _PubSub,
    stdio: StdIOEndpoint,
    keep_alive: float = KEEP_ALIVE_INTERVAL,
    sse_path: str = "/sse",
    message_path: str = "/message",
    cors_origins: Optional[List[str]] = None,
    header_mappings: Optional[NormalizedMappings] = None,
) -> FastAPI:
    """Build FastAPI application with SSE and message endpoints.

    Creates a FastAPI app with SSE streaming endpoint and message posting
    endpoint for bidirectional communication with the stdio subprocess.

    Args:
        pubsub: The publish/subscribe system for message routing.
        stdio: The stdio endpoint for subprocess communication.
        keep_alive: Interval in seconds for keepalive messages. Defaults to KEEP_ALIVE_INTERVAL.
        sse_path: Path for the SSE endpoint. Defaults to "/sse".
        message_path: Path for the message endpoint. Defaults to "/message".
        cors_origins: Optional list of CORS allowed origins.
        header_mappings: Optional mapping of HTTP headers to environment variables.

    Returns:
        FastAPI: The configured FastAPI application.

    Examples:
        >>> pubsub = _PubSub()
        >>> stdio = StdIOEndpoint("cat", pubsub)
        >>> app = _build_fastapi(pubsub, stdio)
        >>> isinstance(app, FastAPI)
        True
        >>> "/sse" in [r.path for r in app.routes]
        True
        >>> "/message" in [r.path for r in app.routes]
        True
        >>> "/healthz" in [r.path for r in app.routes]
        True

        >>> # Test with custom paths
        >>> app2 = _build_fastapi(pubsub, stdio, sse_path="/events", message_path="/send")
        >>> "/events" in [r.path for r in app2.routes]
        True
        >>> "/send" in [r.path for r in app2.routes]
        True

        >>> # Test CORS middleware is added
        >>> app3 = _build_fastapi(pubsub, stdio, cors_origins=["http://example.com"])
        >>> # Check that middleware stack includes CORSMiddleware
        >>> any("CORSMiddleware" in str(m) for m in app3.user_middleware)
        True
    """
    # Standard
    import time as _time  # pylint: disable=import-outside-toplevel

    app = FastAPI()

    # Rate limiter: minimum interval between subprocess restarts (seconds)
    restart_min_interval_secs = 30.0
    last_restart_ts: dict[str, float] = {"t": 0.0}

    def _restart_allowed() -> bool:
        """Return whether enough time elapsed to allow a subprocess restart.

        Returns:
            bool: ``True`` when restart is allowed, otherwise ``False``.
        """
        now = _time.monotonic()
        if now - last_restart_ts["t"] < restart_min_interval_secs:
            return False
        last_restart_ts["t"] = now
        return True

    # Add CORS middleware if origins specified
    if cors_origins:
        app.add_middleware(
            CORSMiddleware,
            allow_origins=cors_origins,
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"],
        )

    # ----- GET /sse ---------------------------------------------------------#
    @app.get(sse_path)
    async def get_sse(request: Request) -> EventSourceResponse:  # noqa: D401
        """Stream subprocess stdout to any number of SSE clients.

        Args:
            request (Request): The incoming ``GET`` request that will be
                upgraded to a Server-Sent Events (SSE) stream.

        Returns:
            EventSourceResponse: A streaming response that forwards JSON-RPC
            messages from the child process and emits periodic ``keepalive``
            frames so that clients and proxies do not time out.
        """
        # Extract environment variables from headers if dynamic env is enabled
        additional_env_vars = {}
        if header_mappings:
            request_headers = dict(request.headers)
            additional_env_vars = extract_env_vars_from_headers(request_headers, header_mappings)

            # Restart stdio endpoint with new environment variables (rate-limited)
            if additional_env_vars and _restart_allowed():
                LOGGER.info(f"Restarting stdio endpoint with {len(additional_env_vars)} environment variables")
                await stdio.stop()
                await stdio.start(additional_env_vars)

        queue = pubsub.subscribe()
        session_id = uuid.uuid4().hex

        async def event_gen() -> AsyncIterator[Dict[str, Any]]:
            """Generate Server-Sent Events for the SSE stream.

            Yields SSE events in the following sequence:
            1. An 'endpoint' event with the message posting URL (required by MCP spec)
            2. An immediate 'keepalive' event to confirm the stream is active
            3. 'message' events containing JSON-RPC responses from the subprocess
            4. Periodic 'keepalive' events to prevent timeouts

            The generator runs until the client disconnects or the server shuts down.
            Automatically unsubscribes from the pubsub system on completion.

            Yields:
                Dict[str, Any]: SSE event dictionaries containing:
                    - event: The event type ('endpoint', 'message', or 'keepalive')
                    - data: The event payload (URL, JSON-RPC message, or empty object)
                    - retry: Retry interval in milliseconds for reconnection

            Examples:
                >>> import asyncio
                >>> async def test_event_gen():
                ...     # This is tested indirectly through the SSE endpoint
                ...     return True
                >>> asyncio.run(test_event_gen())
                True
            """
            # 1️⃣ Mandatory "endpoint" bootstrap required by the MCP spec
            endpoint_url = f"{str(request.base_url).rstrip('/')}{message_path}?session_id={session_id}"
            yield {
                "event": "endpoint",
                "data": endpoint_url,
                "retry": int(keep_alive * 1000),
            }

            # 2️⃣ Immediate keepalive so clients know the stream is alive (if enabled in config)
            if DEFAULT_KEEPALIVE_ENABLED:
                yield {"event": "keepalive", "data": "{}", "retry": keep_alive * 1000}

            try:
                while True:
                    if await request.is_disconnected():
                        break

                    try:
                        timeout = keep_alive if DEFAULT_KEEPALIVE_ENABLED else None
                        msg = await asyncio.wait_for(queue.get(), timeout)
                        yield {"event": "message", "data": msg.rstrip()}
                    except asyncio.TimeoutError:
                        if DEFAULT_KEEPALIVE_ENABLED:
                            yield {
                                "event": "keepalive",
                                "data": "{}",
                                "retry": keep_alive * 1000,
                            }
            finally:
                if pubsub:
                    pubsub.unsubscribe(queue)

        return EventSourceResponse(
            event_gen(),
            headers={
                "Cache-Control": "no-cache",
                "Connection": "keep-alive",
                "X-Accel-Buffering": "no",  # disable proxy buffering
            },
        )

    # ----- POST /message ----------------------------------------------------#
    @app.post(message_path, status_code=status.HTTP_202_ACCEPTED)
    async def post_message(raw: Request, session_id: str | None = None) -> Response:  # noqa: D401
        """Forward a raw JSON-RPC request to the stdio subprocess.

        Args:
            raw (Request): The incoming ``POST`` request whose body contains
                a single JSON-RPC message.
            session_id (str | None): The SSE session identifier that originated
                this back-channel call (present when the client obtained the
                endpoint URL from an ``endpoint`` bootstrap frame).

        Returns:
            Response: ``202 Accepted`` if the payload is forwarded successfully,
            or ``400 Bad Request`` when the body is not valid JSON.
        """
        _ = session_id  # Unused but required for API compatibility

        # Extract environment variables from headers if dynamic env is enabled
        additional_env_vars = {}
        if header_mappings:
            request_headers = dict(raw.headers)
            additional_env_vars = extract_env_vars_from_headers(request_headers, header_mappings)

            # Restart stdio endpoint with new environment variables (rate-limited)
            if additional_env_vars and _restart_allowed():
                LOGGER.info(f"Restarting stdio endpoint with {len(additional_env_vars)} environment variables")
                await stdio.stop()
                await stdio.start(additional_env_vars)
                await asyncio.sleep(0.5)  # Give process time to initialize

        # Ensure stdio endpoint is running
        if not stdio.is_running():
            LOGGER.info("Starting stdio endpoint (was not running)")
            await stdio.start()
            await asyncio.sleep(0.5)  # Give process time to initialize

        payload = await raw.body()
        try:
            orjson.loads(payload)  # validate
        except Exception as exc:  # noqa: BLE001
            return PlainTextResponse(
                f"Invalid JSON payload: {exc}",
                status_code=status.HTTP_400_BAD_REQUEST,
            )
        await stdio.send(payload.decode().rstrip() + "\n")
        return PlainTextResponse("forwarded", status_code=status.HTTP_202_ACCEPTED)

    # ----- Liveness ---------------------------------------------------------#
    @app.get("/healthz")
    async def health() -> Response:  # noqa: D401
        """Health check endpoint.

        Returns:
            Response: A plain text response with "ok" status.
        """
        return PlainTextResponse("ok")

    return app


# ---------------------------------------------------------------------------#
# CLI & orchestration                                                        #
# ---------------------------------------------------------------------------#


def _parse_args(argv: Sequence[str]) -> argparse.Namespace:
    """Parse command line arguments.

    Validates mutually exclusive source options and sets defaults for
    port and logging configuration.

    Args:
        argv: Sequence of command line arguments.

    Returns:
        argparse.Namespace: Parsed command line arguments.

    Raises:
        NotImplementedError: If streamableHttp option is specified.

    Examples:
        >>> args = _parse_args(["--stdio", "cat", "--port", "9000"])
        >>> args.stdio
        'cat'
        >>> args.port
        9000
        >>> args.logLevel
        'info'
        >>> args.host
        '127.0.0.1'
        >>> args.cors is None
        True
        >>> args.oauth2Bearer is None
        True

        >>> # Test default parameters
        >>> args = _parse_args(["--stdio", "cat"])
        >>> args.port
        8000
        >>> args.host
        '127.0.0.1'
        >>> args.logLevel
        'info'

        >>> # Test connect-sse mode
        >>> args = _parse_args(["--connect-sse", "http://example.com/sse"])
        >>> args.connect_sse
        'http://example.com/sse'
        >>> args.stdio is None
        True

        >>> # Test CORS configuration
        >>> args = _parse_args(["--stdio", "cat", "--cors", "https://app.com", "https://web.com"])
        >>> args.cors
        ['https://app.com', 'https://web.com']

        >>> # Test OAuth2 Bearer token
        >>> args = _parse_args(["--connect-sse", "http://example.com", "--oauth2Bearer", "token123"])
        >>> args.oauth2Bearer
        'token123'

        >>> # Test custom host and log level
        >>> args = _parse_args(["--stdio", "cat", "--host", "0.0.0.0", "--logLevel", "debug"])
        >>> args.host
        '0.0.0.0'
        >>> args.logLevel
        'debug'

        >>> # Test expose protocols
        >>> args = _parse_args(["--stdio", "uvx mcp-server-git", "--expose-sse", "--expose-streamable-http"])
        >>> args.stdio
        'uvx mcp-server-git'
        >>> args.expose_sse
        True
        >>> args.expose_streamable_http
        True
        >>> args.stateless
        False
        >>> args.jsonResponse
        False

        >>> # Test new parameters
        >>> args = _parse_args(["--stdio", "cat", "--ssePath", "/events", "--messagePath", "/send", "--keepAlive", "60"])
        >>> args.ssePath
        '/events'
        >>> args.messagePath
        '/send'
        >>> args.keepAlive
        60

        >>> # Test connect-sse with stdio command
        >>> args = _parse_args(["--connect-sse", "http://example.com/sse", "--stdioCommand", "uvx mcp-server-git"])
        >>> args.stdioCommand
        'uvx mcp-server-git'

        >>> # Test connect-sse without stdio command (allowed)
        >>> args = _parse_args(["--connect-sse", "http://example.com/sse"])
        >>> args.stdioCommand is None
        True
    """
    p = argparse.ArgumentParser(
        prog="mcpgateway.translate",
        description="Bridges between different MCP transport protocols: stdio, SSE, and streamable HTTP.",
    )

    # Source/destination options
    p.add_argument("--stdio", help='Local command to run, e.g. "uvx mcp-server-git"')
    p.add_argument("--connect-sse", dest="connect_sse", help="Connect to remote SSE endpoint URL")
    p.add_argument("--connect-streamable-http", dest="connect_streamable_http", help="Connect to remote streamable HTTP endpoint URL")
    p.add_argument("--grpc", type=str, help="gRPC server target (host:port) to expose")
    p.add_argument("--connect-grpc", type=str, help="Remote gRPC endpoint to connect to")

    # Protocol exposure options (can be combined)
    p.add_argument("--expose-sse", action="store_true", help="Expose via SSE protocol (endpoints: /sse and /message)")
    p.add_argument("--expose-streamable-http", action="store_true", help="Expose via streamable HTTP protocol (endpoint: /mcp)")

    # gRPC configuration options
    p.add_argument("--grpc-tls", action="store_true", help="Enable TLS for gRPC connection")
    p.add_argument("--grpc-cert", type=str, help="Path to TLS certificate for gRPC")
    p.add_argument("--grpc-key", type=str, help="Path to TLS key for gRPC")
    p.add_argument("--grpc-metadata", action="append", help="gRPC metadata (KEY=VALUE, repeatable)")

    p.add_argument("--port", type=int, default=8000, help="HTTP port to bind")
    p.add_argument("--host", default="127.0.0.1", help="Host interface to bind (default: 127.0.0.1)")
    p.add_argument(
        "--logLevel",
        default="info",
        choices=["debug", "info", "warning", "error", "critical"],
        help="Log level",
    )
    p.add_argument(
        "--cors",
        nargs="*",
        help="CORS allowed origins (e.g., --cors https://app.example.com)",
    )
    p.add_argument(
        "--oauth2Bearer",
        help="OAuth2 Bearer token for authentication",
    )

    # New configuration options
    p.add_argument(
        "--ssePath",
        default="/sse",
        help="SSE endpoint path (default: /sse)",
    )
    p.add_argument(
        "--messagePath",
        default="/message",
        help="Message endpoint path (default: /message)",
    )
    p.add_argument(
        "--keepAlive",
        type=int,
        default=KEEP_ALIVE_INTERVAL,
        help=f"Keep-alive interval in seconds (default: {KEEP_ALIVE_INTERVAL})",
    )

    # For SSE to stdio mode
    p.add_argument(
        "--stdioCommand",
        help="Command to run when bridging SSE/streamableHttp to stdio (optional with --sse or --streamableHttp)",
    )

    # Dynamic environment variable injection
    p.add_argument("--enable-dynamic-env", action="store_true", help="Enable dynamic environment variable injection from HTTP headers")
    p.add_argument(
        "--header-to-env",
        action="append",
        default=[],
        help="Map HTTP header to environment variable (format: HEADER=ENV_VAR, can be used multiple times). Case-insensitive duplicates are rejected (e.g., Authorization and authorization cannot both be mapped).",
    )

    # For streamable HTTP mode
    p.add_argument(
        "--stateless",
        action="store_true",
        help="Use stateless mode for streamable HTTP (default: False)",
    )
    p.add_argument(
        "--jsonResponse",
        action="store_true",
        help="Return JSON responses instead of SSE streams for streamable HTTP (default: False)",
    )

    args = p.parse_args(argv)
    # streamableHttp is now supported, no need to raise NotImplementedError
    return args


async def _run_stdio_to_sse(
    cmd: str,
    port: int,
    log_level: str = "info",
    cors: Optional[List[str]] = None,
    host: str = "127.0.0.1",
    sse_path: str = "/sse",
    message_path: str = "/message",
    keep_alive: float = KEEP_ALIVE_INTERVAL,
    header_mappings: Optional[NormalizedMappings] = None,
) -> None:
    """Run stdio to SSE bridge.

    Starts a subprocess and exposes it via HTTP/SSE endpoints. Handles graceful
    shutdown on SIGINT/SIGTERM.

    Args:
        cmd: The command to run as a stdio subprocess.
        port: The port to bind the HTTP server to.
        log_level: The logging level to use. Defaults to "info".
        cors: Optional list of CORS allowed origins.
        host: The host interface to bind to. Defaults to "127.0.0.1" for security.
        sse_path: Path for the SSE endpoint. Defaults to "/sse".
        message_path: Path for the message endpoint. Defaults to "/message".
        keep_alive: Keep-alive interval in seconds. Defaults to KEEP_ALIVE_INTERVAL.
        header_mappings: Optional mapping of HTTP headers to environment variables.

    Examples:
        >>> import asyncio # doctest: +SKIP
        >>> async def test_run(): # doctest: +SKIP
        ...     await _run_stdio_to_sse("cat", 9000)  # doctest: +SKIP
        ...     return True
        >>> asyncio.run(test_run()) # doctest: +SKIP
        True
    """
    pubsub = _PubSub()
    stdio = StdIOEndpoint(cmd, pubsub, header_mappings=header_mappings)
    await stdio.start()

    app = _build_fastapi(pubsub, stdio, keep_alive=keep_alive, sse_path=sse_path, message_path=message_path, cors_origins=cors, header_mappings=header_mappings)
    config = uvicorn.Config(
        app,
        host=host,  # Changed from hardcoded "0.0.0.0"
        port=port,
        log_level=log_level,
        lifespan="off",
    )
    uvicorn_server = uvicorn.Server(config)

    shutting_down = asyncio.Event()  # 🔄 make shutdown idempotent

    async def _shutdown() -> None:
        """Handle graceful shutdown of the stdio bridge.

        Performs shutdown operations in the correct order:
        1. Sets a flag to prevent multiple shutdown attempts
        2. Stops the stdio subprocess
        3. Shuts down the HTTP server

        This function is idempotent - multiple calls will only execute
        the shutdown sequence once.

        Examples:
            >>> import asyncio
            >>> async def test_shutdown():
            ...     # Shutdown is tested as part of the main run flow
            ...     return True
            >>> asyncio.run(test_shutdown())
            True
        """
        if shutting_down.is_set():
            return
        shutting_down.set()
        LOGGER.info("Shutting down ...")
        await stdio.stop()
        # Graceful shutdown by setting the shutdown event
        # Use getattr to safely access should_exit attribute
        setattr(uvicorn_server, "should_exit", getattr(uvicorn_server, "should_exit", False) or True)

    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        with suppress(NotImplementedError):  # Windows lacks add_signal_handler
            loop.add_signal_handler(sig, lambda *_: asyncio.create_task(_shutdown()))

    LOGGER.info(f"Bridge ready → http://{host}:{port}{sse_path}")
    await uvicorn_server.serve()
    await _shutdown()  # final cleanup


async def _run_sse_to_stdio(url: str, oauth2_bearer: Optional[str] = None, timeout: float = 30.0, stdio_command: Optional[str] = None, max_retries: int = 5, initial_retry_delay: float = 1.0) -> None:
    """Run SSE to stdio bridge.

    Connects to a remote SSE endpoint and bridges it to local stdio.
    Implements proper bidirectional message flow with error handling and retries.

    Args:
        url: The SSE endpoint URL to connect to.
        oauth2_bearer: Optional OAuth2 bearer token for authentication. Defaults to None.
        timeout: HTTP client timeout in seconds. Defaults to 30.0.
        stdio_command: Optional command to run for local stdio processing.
            If not provided, will simply print SSE messages to stdout.
        max_retries: Maximum number of connection retry attempts. Defaults to 5.
        initial_retry_delay: Initial delay between retries in seconds. Defaults to 1.0.

    Raises:
        ImportError: If httpx package is not available.
        RuntimeError: If the subprocess fails to create stdin/stdout pipes.
        Exception: For any unexpected error in SSE stream processing.

    Examples:
        >>> import asyncio
        >>> async def test_sse():
        ...     try:
        ...         await _run_sse_to_stdio("http://example.com/sse", None)  # doctest: +SKIP
        ...     except ImportError as e:
        ...         return "httpx" in str(e)
        >>> asyncio.run(test_sse())  # Would return True if httpx not installed # doctest: +SKIP
    """
    if not httpx:
        raise ImportError("httpx package is required for SSE to stdio bridging")

    headers = {}
    if oauth2_bearer:
        headers["Authorization"] = f"Bearer {oauth2_bearer}"

    # If no stdio command provided, use simple mode (just print to stdout)
    if not stdio_command:
        LOGGER.warning("No --stdioCommand provided, running in simple mode (SSE to stdout only)")
        # First-Party
        from mcpgateway.services.http_client_service import get_isolated_http_client  # pylint: disable=import-outside-toplevel

        async with get_isolated_http_client(timeout=timeout, headers=headers, verify=DEFAULT_SSL_VERIFY, connect_timeout=10.0, write_timeout=timeout, pool_timeout=timeout) as client:
            await _simple_sse_pump(client, url, max_retries, initial_retry_delay)
        return

    # Start the stdio subprocess
    process = await asyncio.create_subprocess_exec(
        *shlex.split(stdio_command),
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=sys.stderr,
    )

    if not process.stdin or not process.stdout:
        raise RuntimeError(f"Failed to create subprocess with stdin/stdout pipes for command: {stdio_command}")

    # Store the message endpoint URL once received
    message_endpoint: Optional[str] = None

    async def read_stdout(client: httpx.AsyncClient) -> None:
        """Read lines from subprocess stdout and POST to message endpoint.

        Continuously reads JSON-RPC requests from the subprocess stdout
        and POSTs them to the remote message endpoint obtained from the
        SSE stream's endpoint event.

        Args:
            client: The HTTP client to use for POSTing messages.

        Raises:
            RuntimeError: If the process stdout stream is not available.

        Examples:
            >>> import asyncio
            >>> async def test_read():
            ...     # This is tested as part of the SSE to stdio flow
            ...     return True
            >>> asyncio.run(test_read())
            True
        """
        if not process.stdout:
            raise RuntimeError("Process stdout not available")

        while True:
            if not process.stdout:
                raise RuntimeError("Process stdout not available")
            line = await process.stdout.readline()
            if not line:
                break

            text = line.decode().strip()
            if not text:
                continue

            LOGGER.debug(f"← stdio: {text}")

            # Wait for endpoint URL if not yet received
            retry_count = 0
            while not message_endpoint and retry_count < 30:  # 30 second timeout
                await asyncio.sleep(1)
                retry_count += 1

            if not message_endpoint:
                LOGGER.error("No message endpoint received from SSE stream")
                continue

            # POST the JSON-RPC request to the message endpoint
            try:
                response = await client.post(message_endpoint, content=text, headers={"Content-Type": "application/json"})
                if response.status_code != 202:
                    LOGGER.warning(f"Message endpoint returned {response.status_code}: {response.text}")
            except Exception as e:
                LOGGER.error(f"Failed to POST to message endpoint: {e}")

    async def pump_sse_to_stdio(client: httpx.AsyncClient) -> None:
        """Stream SSE data from remote endpoint to subprocess stdin.

        Connects to the remote SSE endpoint with retry logic and forwards
        message events to the subprocess stdin. Properly parses SSE events
        and handles endpoint, message, and keepalive event types.

        Args:
            client: The HTTP client to use for SSE streaming.

        Raises:
            HTTPStatusError: If the SSE endpoint returns a non-200 status code.
            Exception: For unexpected errors in SSE stream processing.

        Examples:
            >>> import asyncio
            >>> async def test_pump():
            ...     # This is tested as part of the SSE to stdio flow
            ...     return True
            >>> asyncio.run(test_pump())
            True
        """
        nonlocal message_endpoint
        retry_delay = initial_retry_delay
        retry_count = 0

        while retry_count < max_retries:
            try:
                LOGGER.info(f"Connecting to SSE endpoint: {url}")

                async with client.stream("GET", url) as response:
                    # Check status code if available (real httpx response)
                    if hasattr(response, "status_code") and response.status_code != 200:
                        if httpx:
                            raise httpx.HTTPStatusError(f"SSE endpoint returned {response.status_code}", request=response.request, response=response)
                        raise Exception(f"SSE endpoint returned {response.status_code}")

                    # Reset retry counter on successful connection
                    retry_count = 0
                    retry_delay = initial_retry_delay
                    current_event: Optional[SSEEvent] = None

                    async for line in response.aiter_lines():
                        event, is_complete = SSEEvent.parse_sse_line(line, current_event)
                        current_event = event

                        if is_complete and current_event:
                            LOGGER.debug(f"SSE event: {current_event.event} - {current_event.data[:100]}...")

                            if current_event.event == "endpoint":
                                # Store the message endpoint URL
                                message_endpoint = current_event.data
                                LOGGER.info(f"Received message endpoint: {message_endpoint}")

                            elif current_event.event == "message":
                                # Forward JSON-RPC responses to stdio
                                if process.stdin:
                                    process.stdin.write((current_event.data + "\n").encode())
                                    await process.stdin.drain()
                                    LOGGER.debug(f"→ stdio: {current_event.data}")

                            elif current_event.event == "keepalive":
                                # Log keepalive but don't forward
                                LOGGER.debug("Received keepalive")

                            # Reset for next event
                            current_event = None

            except Exception as e:
                # Check if it's one of the expected httpx exceptions
                if httpx and isinstance(e, (httpx.ConnectError, httpx.HTTPStatusError, httpx.ReadTimeout)):
                    retry_count += 1
                    if retry_count >= max_retries:
                        LOGGER.error(f"Max retries ({max_retries}) exceeded. Giving up.")
                        raise

                    LOGGER.warning(f"Connection error: {e}. Retrying in {retry_delay}s... (attempt {retry_count}/{max_retries})")
                    await asyncio.sleep(retry_delay)
                    retry_delay = min(retry_delay * 2, 30)  # Exponential backoff, max 30s
                else:
                    LOGGER.error(f"Unexpected error in SSE stream: {e}")
                    raise

    # Run both tasks concurrently
    # First-Party
    from mcpgateway.services.http_client_service import get_isolated_http_client  # pylint: disable=import-outside-toplevel

    async with get_isolated_http_client(timeout=timeout, headers=headers, verify=DEFAULT_SSL_VERIFY, connect_timeout=10.0, write_timeout=timeout, pool_timeout=timeout) as client:
        try:
            await asyncio.gather(read_stdout(client), pump_sse_to_stdio(client))
        except Exception as e:
            LOGGER.error(f"Bridge error: {e}")
            raise
        finally:
            # Clean up subprocess
            if process.returncode is None:
                process.terminate()
                with suppress(asyncio.TimeoutError):
                    await asyncio.wait_for(process.wait(), timeout=5)


async def _run_stdio_to_streamable_http(
    cmd: str,
    port: int,
    log_level: str = "info",
    cors: Optional[List[str]] = None,
    host: str = "127.0.0.1",
    stateless: bool = False,
    json_response: bool = False,
) -> None:
    """Run stdio to streamable HTTP bridge.

    Starts a subprocess and exposes it via streamable HTTP endpoint. Handles graceful
    shutdown on SIGINT/SIGTERM.

    Args:
        cmd: The command to run as a stdio subprocess.
        port: The port to bind the HTTP server to.
        log_level: The logging level to use. Defaults to "info".
        cors: Optional list of CORS allowed origins.
        host: The host interface to bind to. Defaults to "127.0.0.1" for security.
        stateless: Whether to use stateless mode for streamable HTTP. Defaults to False.
        json_response: Whether to return JSON responses instead of SSE streams. Defaults to False.

    Raises:
        ImportError: If MCP server components are not available.
        RuntimeError: If subprocess fails to create stdin/stdout pipes.

    Examples:
        >>> import asyncio
        >>> async def test_streamable_http():
        ...     # Would start a real subprocess and HTTP server
        ...     cmd = "echo hello"
        ...     port = 9000
        ...     # This would normally run the server
        ...     return True
        >>> asyncio.run(test_streamable_http())
        True
    """
    # MCP components are available, proceed with setup

    LOGGER.info(f"Starting stdio to streamable HTTP bridge for command: {cmd}")

    # Create a simple MCP server that will proxy to stdio subprocess
    mcp_server = MCPServer(name="stdio-proxy")

    # Create subprocess for stdio communication
    process = await asyncio.create_subprocess_exec(
        *shlex.split(cmd),
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=sys.stderr,
    )

    if not process.stdin or not process.stdout:
        raise RuntimeError(f"Failed to create subprocess with stdin/stdout pipes for command: {cmd}")

    # Set up the streamable HTTP session manager with the server
    session_manager = StreamableHTTPSessionManager(
        app=mcp_server,
        stateless=stateless,
        json_response=json_response,
    )

    # Create Starlette app to host the streamable HTTP endpoint
    async def handle_mcp(request: Request) -> None:
        """Handle MCP requests via streamable HTTP.

        Args:
            request: The incoming HTTP request from Starlette.

        Examples:
            >>> async def test_handle():
            ...     # Mock request handling
            ...     class MockRequest:
            ...         scope = {"type": "http"}
            ...         async def receive(self): return {}
            ...         async def send(self, msg): return None
            ...     req = MockRequest()
            ...     # Would handle the request via session manager
            ...     return req is not None
            >>> import asyncio
            >>> asyncio.run(test_handle())
            True
        """
        # The session manager handles all the protocol details - Note: I don't like accessing _send directly -JPS
        await session_manager.handle_request(request.scope, request.receive, request._send)  # pylint: disable=W0212

    routes = [
        Route("/mcp", handle_mcp, methods=["GET", "POST"]),
        Route("/healthz", lambda request: PlainTextResponse("ok"), methods=["GET"]),
    ]

    app = Starlette(routes=routes)

    # Add CORS middleware if specified
    if cors:
        app.add_middleware(
            CORSMiddleware,
            allow_origins=cors,
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"],
        )

    # Run the server with Uvicorn
    config = uvicorn.Config(
        app,
        host=host,
        port=port,
        log_level=log_level,
        lifespan="off",
    )
    uvicorn_server = uvicorn.Server(config)

    shutting_down = asyncio.Event()

    async def _shutdown() -> None:
        """Handle graceful shutdown of the streamable HTTP bridge."""
        if shutting_down.is_set():
            return
        shutting_down.set()
        LOGGER.info("Shutting down streamable HTTP bridge...")
        if process.returncode is None:
            process.terminate()
            with suppress(asyncio.TimeoutError):
                await asyncio.wait_for(process.wait(), 5)
        # Graceful shutdown by setting the shutdown event
        # Use getattr to safely access should_exit attribute
        setattr(uvicorn_server, "should_exit", getattr(uvicorn_server, "should_exit", False) or True)

    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        with suppress(NotImplementedError):  # Windows lacks add_signal_handler
            loop.add_signal_handler(sig, lambda *_: asyncio.create_task(_shutdown()))

    # Pump messages between stdio and HTTP
    async def pump_stdio_to_http() -> None:
        """Forward messages from subprocess stdout to HTTP responses.

        Examples:
            >>> async def test():
            ...     # This would pump messages in real usage
            ...     return True
            >>> import asyncio
            >>> asyncio.run(test())
            True
        """
        while True:
            try:
                if not process.stdout:
                    raise RuntimeError("Process stdout not available")
                line = await process.stdout.readline()
                if not line:
                    break
                # The session manager will handle routing to appropriate HTTP responses
                # This would need proper integration with session_manager's internal queue
                LOGGER.debug(f"Received from subprocess: {line.decode().strip()}")
            except Exception as e:
                LOGGER.error(f"Error reading from subprocess: {e}")
                break

    async def pump_http_to_stdio(data: str) -> None:
        """Forward HTTP requests to subprocess stdin.

        Args:
            data: The data string to send to subprocess stdin.

        Examples:
            >>> async def test_pump():
            ...     # Would pump data to subprocess
            ...     data = '{"method": "test"}'
            ...     # In real use, would write to process.stdin
            ...     return len(data) > 0
            >>> import asyncio
            >>> asyncio.run(test_pump())
            True
        """
        if not process.stdin:
            raise RuntimeError("Process stdin not available")
        process.stdin.write(data.encode() + b"\n")
        await process.stdin.drain()

    # Note: pump_http_to_stdio will be used when stdio-to-HTTP bridge is fully implemented
    _ = pump_http_to_stdio

    # Start the pump task
    pump_task = asyncio.create_task(pump_stdio_to_http())

    try:
        LOGGER.info(f"Streamable HTTP bridge ready → http://{host}:{port}/mcp")
        await uvicorn_server.serve()
    finally:
        pump_task.cancel()
        await _shutdown()


async def _run_streamable_http_to_stdio(
    url: str,
    oauth2_bearer: Optional[str] = None,
    timeout: float = 30.0,
    stdio_command: Optional[str] = None,
    max_retries: int = 5,
    initial_retry_delay: float = 1.0,
) -> None:
    """Run streamable HTTP to stdio bridge.

    Connects to a remote streamable HTTP endpoint and bridges it to local stdio.
    Implements proper bidirectional message flow with error handling and retries.

    Args:
        url: The streamable HTTP endpoint URL to connect to.
        oauth2_bearer: Optional OAuth2 bearer token for authentication. Defaults to None.
        timeout: HTTP client timeout in seconds. Defaults to 30.0.
        stdio_command: Optional command to run for local stdio processing.
            If not provided, will simply print messages to stdout.
        max_retries: Maximum number of connection retry attempts. Defaults to 5.
        initial_retry_delay: Initial delay between retries in seconds. Defaults to 1.0.

    Raises:
        ImportError: If httpx package is not available.
        RuntimeError: If the subprocess fails to create stdin/stdout pipes.
        Exception: For any unexpected error during bridging operations.
    """
    if not httpx:
        raise ImportError("httpx package is required for streamable HTTP to stdio bridging")

    headers = {}
    if oauth2_bearer:
        headers["Authorization"] = f"Bearer {oauth2_bearer}"

    # Ensure URL ends with /mcp if not already
    if not url.endswith("/mcp"):
        url = url.rstrip("/") + "/mcp"

    # If no stdio command provided, use simple mode (just print to stdout)
    if not stdio_command:
        LOGGER.warning("No --stdioCommand provided, running in simple mode (streamable HTTP to stdout only)")
        # First-Party
        from mcpgateway.services.http_client_service import get_isolated_http_client  # pylint: disable=import-outside-toplevel

        async with get_isolated_http_client(timeout=timeout, headers=headers, verify=DEFAULT_SSL_VERIFY, connect_timeout=10.0, write_timeout=timeout, pool_timeout=timeout) as client:
            await _simple_streamable_http_pump(client, url, max_retries, initial_retry_delay)
        return

    # Start the stdio subprocess
    process = await asyncio.create_subprocess_exec(
        *shlex.split(stdio_command),
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=sys.stderr,
    )

    if not process.stdin or not process.stdout:
        raise RuntimeError(f"Failed to create subprocess with stdin/stdout pipes for command: {stdio_command}")

    async def read_stdout(client: httpx.AsyncClient) -> None:
        """Read lines from subprocess stdout and POST to streamable HTTP endpoint.

        Args:
            client: The HTTP client to use for POSTing messages.

        Raises:
            RuntimeError: If the process stdout stream is not available.
        """
        if not process.stdout:
            raise RuntimeError("Process stdout not available")

        while True:
            if not process.stdout:
                raise RuntimeError("Process stdout not available")
            line = await process.stdout.readline()
            if not line:
                break

            text = line.decode().strip()
            if not text:
                continue

            LOGGER.debug(f"← stdio: {text}")

            # POST the JSON-RPC request to the streamable HTTP endpoint
            try:
                if CONTENT_TYPE == "application/x-www-form-urlencoded":
                    # If text is JSON, parse and encode as form
                    try:
                        payload = orjson.loads(text)
                        body = urlencode(payload)
                    except Exception:
                        body = text
                    response = await client.post(url, content=body, headers=headers)
                else:
                    response = await client.post(url, content=text, headers=headers)
                if response.status_code == 200:
                    # Handle JSON response
                    response_data = response.text
                    if response_data and process.stdin:
                        process.stdin.write((response_data + "\n").encode())
                        await process.stdin.drain()
                        LOGGER.debug(f"→ stdio: {response_data}")
                else:
                    LOGGER.warning(f"Streamable HTTP endpoint returned {response.status_code}: {response.text}")
            except Exception as e:
                LOGGER.error(f"Failed to POST to streamable HTTP endpoint: {e}")

    async def pump_streamable_http_to_stdio(client: httpx.AsyncClient) -> None:
        """Stream data from remote streamable HTTP endpoint to subprocess stdin.

        Args:
            client: The HTTP client to use for streamable HTTP streaming.

        Raises:
            httpx.HTTPStatusError: If the streamable HTTP endpoint returns a non-200 status code.
            Exception: For unexpected errors in streamable HTTP stream processing.
        """
        retry_delay = initial_retry_delay
        retry_count = 0

        while retry_count < max_retries:
            try:
                LOGGER.info(f"Connecting to streamable HTTP endpoint: {url}")

                # For streamable HTTP, we need to handle both SSE streams and JSON responses
                # Try SSE first (for stateful sessions or when SSE is preferred)
                async with client.stream("GET", url, headers={"Accept": "text/event-stream"}) as response:
                    if response.status_code != 200:
                        if httpx:
                            raise httpx.HTTPStatusError(f"Streamable HTTP endpoint returned {response.status_code}", request=response.request, response=response)
                        raise Exception(f"Streamable HTTP endpoint returned {response.status_code}")

                    # Reset retry counter on successful connection
                    retry_count = 0
                    retry_delay = initial_retry_delay

                    async for line in response.aiter_lines():
                        if line.startswith("data:"):
                            data = line[5:]
                            if data.startswith(" "):
                                data = data[1:]
                            if data and process.stdin:
                                process.stdin.write((data + "\n").encode())
                                await process.stdin.drain()
                                LOGGER.debug(f"→ stdio: {data}")

            except Exception as e:
                if httpx and isinstance(e, (httpx.ConnectError, httpx.HTTPStatusError, httpx.ReadTimeout)):
                    retry_count += 1
                    if retry_count >= max_retries:
                        LOGGER.error(f"Max retries ({max_retries}) exceeded. Giving up.")
                        raise

                    LOGGER.warning(f"Connection error: {e}. Retrying in {retry_delay}s... (attempt {retry_count}/{max_retries})")
                    await asyncio.sleep(retry_delay)
                    retry_delay = min(retry_delay * 2, 30)  # Exponential backoff, max 30s
                else:
                    LOGGER.error(f"Unexpected error in streamable HTTP stream: {e}")
                    raise

    # Run both tasks concurrently
    # First-Party
    from mcpgateway.services.http_client_service import get_isolated_http_client  # pylint: disable=import-outside-toplevel

    async with get_isolated_http_client(timeout=timeout, headers=headers, verify=DEFAULT_SSL_VERIFY, connect_timeout=10.0, write_timeout=timeout, pool_timeout=timeout) as client:
        try:
            await asyncio.gather(read_stdout(client), pump_streamable_http_to_stdio(client))
        except Exception as e:
            LOGGER.error(f"Bridge error: {e}")
            raise
        finally:
            # Clean up subprocess
            if process.returncode is None:
                process.terminate()
                with suppress(asyncio.TimeoutError):
                    await asyncio.wait_for(process.wait(), timeout=5)


async def _simple_streamable_http_pump(client: "Any", url: str, max_retries: int, initial_retry_delay: float) -> None:
    """Simple streamable HTTP pump that just prints messages to stdout.

    Used when no stdio command is provided to bridge streamable HTTP to stdout directly.

    Args:
        client: The HTTP client to use for streamable HTTP streaming.
        url: The streamable HTTP endpoint URL to connect to.
        max_retries: Maximum number of connection retry attempts.
        initial_retry_delay: Initial delay between retries in seconds.

    Raises:
        Exception: For unexpected errors in streamable HTTP stream processing including
            HTTPStatusError if the endpoint returns a non-200 status code.
    """
    retry_delay = initial_retry_delay
    retry_count = 0

    while retry_count < max_retries:
        try:
            LOGGER.info(f"Connecting to streamable HTTP endpoint: {url}")

            # Try to get SSE stream
            async with client.stream("GET", url, headers={"Accept": "text/event-stream"}) as response:
                if response.status_code != 200:
                    if httpx:
                        raise httpx.HTTPStatusError(f"Streamable HTTP endpoint returned {response.status_code}", request=response.request, response=response)
                    raise Exception(f"Streamable HTTP endpoint returned {response.status_code}")

                # Reset retry counter on successful connection
                retry_count = 0
                retry_delay = initial_retry_delay

                async for line in response.aiter_lines():
                    if line.startswith("data:"):
                        data = line[5:]
                        if data.startswith(" "):
                            data = data[1:]
                        if data:
                            print(data)
                            LOGGER.debug(f"Received: {data}")

        except Exception as e:
            if httpx and isinstance(e, (httpx.ConnectError, httpx.HTTPStatusError, httpx.ReadTimeout)):
                retry_count += 1
                if retry_count >= max_retries:
                    LOGGER.error(f"Max retries ({max_retries}) exceeded. Giving up.")
                    raise

                LOGGER.warning(f"Connection error: {e}. Retrying in {retry_delay}s... (attempt {retry_count}/{max_retries})")
                await asyncio.sleep(retry_delay)
                retry_delay = min(retry_delay * 2, 30)  # Exponential backoff, max 30s
            else:
                LOGGER.error(f"Unexpected error in streamable HTTP stream: {e}")
                raise


async def _run_multi_protocol_server(  # pylint: disable=too-many-positional-arguments
    cmd: str,
    port: int,
    log_level: str = "info",
    cors: Optional[List[str]] = None,
    host: str = "127.0.0.1",
    expose_sse: bool = False,
    expose_streamable_http: bool = False,
    sse_path: str = "/sse",
    message_path: str = "/message",
    keep_alive: float = KEEP_ALIVE_INTERVAL,
    stateless: bool = False,
    json_response: bool = False,
    header_mappings: Optional[NormalizedMappings] = None,
) -> None:
    """Run a stdio server and expose it via multiple protocols simultaneously.

    Args:
        cmd: The command to run as a stdio subprocess.
        port: The port to bind the HTTP server to.
        log_level: The logging level to use. Defaults to "info".
        cors: Optional list of CORS allowed origins.
        host: The host interface to bind to. Defaults to "127.0.0.1".
        expose_sse: Whether to expose via SSE protocol.
        expose_streamable_http: Whether to expose via streamable HTTP protocol.
        sse_path: Path for SSE endpoint. Defaults to "/sse".
        message_path: Path for message endpoint. Defaults to "/message".
        keep_alive: Keep-alive interval for SSE. Defaults to KEEP_ALIVE_INTERVAL.
        stateless: Whether to use stateless mode for streamable HTTP.
        json_response: Whether to return JSON responses for streamable HTTP.
        header_mappings: Optional mapping of HTTP headers to environment variables.
    """
    LOGGER.info(f"Starting multi-protocol server for command: {cmd}")
    LOGGER.info(f"Protocols: SSE={expose_sse}, StreamableHTTP={expose_streamable_http}")

    # Create a shared pubsub whenever either protocol needs stdout observations
    pubsub = _PubSub() if (expose_sse or expose_streamable_http) else None

    # Create the stdio endpoint
    stdio = StdIOEndpoint(cmd, pubsub, header_mappings=header_mappings) if (expose_sse or expose_streamable_http) and pubsub else None

    # Create fastapi app and middleware
    app = FastAPI()

    # Add CORS middleware if specified
    if cors:
        app.add_middleware(
            CORSMiddleware,
            allow_origins=cors,
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"],
        )

    # Start stdio if at least one transport requires it
    if stdio:
        await stdio.start()

    # SSE endpoints
    if expose_sse and stdio and pubsub:

        @app.get(sse_path)
        async def get_sse(request: Request) -> EventSourceResponse:
            """SSE endpoint.

            Args:
                request: The incoming HTTP request.

            Returns:
                EventSourceResponse: Server-sent events stream.
            """
            if not pubsub:
                raise RuntimeError("PubSub not available")

            # Extract environment variables from headers if dynamic env is enabled
            additional_env_vars = {}
            if header_mappings and stdio:
                request_headers = dict(request.headers)
                additional_env_vars = extract_env_vars_from_headers(request_headers, header_mappings)

                # Restart stdio endpoint with new environment variables
                if additional_env_vars:
                    LOGGER.info(f"Restarting stdio endpoint with {len(additional_env_vars)} environment variables")
                    await stdio.stop()  # Stop existing process
                    await stdio.start(additional_env_vars)  # Start with new env vars

            queue = pubsub.subscribe()
            session_id = uuid.uuid4().hex

            async def event_gen() -> AsyncIterator[Dict[str, Any]]:
                """Generate SSE events for the client.

                Yields:
                    Dict[str, Any]: SSE event data with event type and payload.
                """
                endpoint_url = f"{str(request.base_url).rstrip('/')}{message_path}?session_id={session_id}"
                yield {
                    "event": "endpoint",
                    "data": endpoint_url,
                    "retry": int(keep_alive * 1000),
                }

                if DEFAULT_KEEPALIVE_ENABLED:
                    yield {"event": "keepalive", "data": "{}", "retry": keep_alive * 1000}

                try:
                    while True:
                        if await request.is_disconnected():
                            break

                        try:
                            timeout = keep_alive if DEFAULT_KEEPALIVE_ENABLED else None
                            msg = await asyncio.wait_for(queue.get(), timeout)
                            yield {"event": "message", "data": msg.rstrip()}
                        except asyncio.TimeoutError:
                            if DEFAULT_KEEPALIVE_ENABLED:
                                yield {
                                    "event": "keepalive",
                                    "data": "{}",
                                    "retry": keep_alive * 1000,
                                }
                finally:
                    if pubsub:
                        pubsub.unsubscribe(queue)

            return EventSourceResponse(
                event_gen(),
                headers={
                    "Cache-Control": "no-cache",
                    "Connection": "keep-alive",
                    "X-Accel-Buffering": "no",
                },
            )

        @app.post(message_path, status_code=status.HTTP_202_ACCEPTED)
        async def post_message(raw: Request, session_id: str | None = None) -> Response:
            """Message endpoint for SSE.

            Args:
                raw: The incoming HTTP request.
                session_id: Optional session ID for correlation.

            Returns:
                Response: Acknowledgement of message receipt.
            """
            _ = session_id

            # Extract environment variables from headers if dynamic env is enabled
            additional_env_vars = {}
            if header_mappings and stdio:
                request_headers = dict(raw.headers)
                additional_env_vars = extract_env_vars_from_headers(request_headers, header_mappings)

                # Only restart if we have new environment variables
                if additional_env_vars:
                    LOGGER.info(f"Restarting stdio endpoint with {len(additional_env_vars)} environment variables")
                    await stdio.stop()  # Stop existing process
                    await stdio.start(additional_env_vars)  # Start with new env vars
                    await asyncio.sleep(0.5)  # Give process time to initialize

            # Ensure stdio endpoint is running
            if stdio and not stdio.is_running():
                LOGGER.info("Starting stdio endpoint (was not running)")
                await stdio.start()
                await asyncio.sleep(0.5)  # Give process time to initialize

            payload = await raw.body()
            try:
                orjson.loads(payload)
            except Exception as exc:
                return PlainTextResponse(
                    f"Invalid JSON payload: {exc}",
                    status_code=status.HTTP_400_BAD_REQUEST,
                )
            if not stdio:
                raise RuntimeError("Stdio endpoint not available")
            await stdio.send(payload.decode().rstrip() + "\n")
            return PlainTextResponse("forwarded", status_code=status.HTTP_202_ACCEPTED)

    # Add health check
    @app.get("/healthz")
    async def health() -> Response:
        """Health check endpoint.

        Returns:
            Response: Health status response.
        """
        return PlainTextResponse("ok")

    # Streamable HTTP support
    streamable_server = None
    streamable_manager = None
    streamable_context = None

    # Keep a reference to the original FastAPI app so we can wrap it with an ASGI
    # layer that delegates `/mcp` scopes to the StreamableHTTPSessionManager if present.
    original_app = app

    if expose_streamable_http:
        # Create an MCP server instance
        streamable_server = MCPServer("stdio-proxy")

        # Set up the streamable HTTP session manager
        streamable_manager = StreamableHTTPSessionManager(
            app=streamable_server,
            stateless=stateless,
            json_response=json_response,
        )

        # Register POST /mcp on the FastAPI app as the canonical client->server POST
        # path for Streamable HTTP. This forwards JSON-RPC notifications/requests to stdio.
        @original_app.post("/mcp")
        async def mcp_post(request: Request) -> Response:
            """
            Handles POST requests to the `/mcp` endpoint, forwarding JSON payloads to stdio
            and optionally waiting for a correlated response.

            The request body is expected to be a JSON object or newline-delimited JSON.
            If the JSON includes an "id" field, the function attempts to match it with
            a response from stdio using a pubsub queue, within a timeout period.

            Args:
                request (Request): The incoming FastAPI request containing the JSON payload.

            Returns:
                Response: A FastAPI Response object.
                    - 200 OK with matched JSON response if correlation succeeds.
                    - 202 Accepted if no matching response is received in time or for notifications.
                    - 400 Bad Request if the payload is not valid JSON.

            Example:
                >>> import httpx
                >>> response = httpx.post("http://localhost:8000/mcp", json={"id": 123, "method": "ping"})
                >>> response.status_code in (200, 202)
                True
                >>> response.text  # May be the matched JSON or "accepted"
                '{"id": 123, "result": "pong"}'  # or "accepted"
            """
            # Read and validate JSON
            body = await request.body()
            try:
                obj = orjson.loads(body)
            except Exception as exc:
                return PlainTextResponse(f"Invalid JSON payload: {exc}", status_code=status.HTTP_400_BAD_REQUEST)

            # Forward raw newline-delimited JSON to stdio
            if not stdio:
                raise RuntimeError("Stdio endpoint not available")
            await stdio.send(body.decode().rstrip() + "\n")

            # If it's a request (has an id) -> attempt to correlate response from stdio
            if isinstance(obj, dict) and "id" in obj:
                if not pubsub:
                    return PlainTextResponse("accepted", status_code=status.HTTP_202_ACCEPTED)

                queue = pubsub.subscribe()
                try:
                    timeout = 10.0  # seconds; tuneable
                    deadline = asyncio.get_event_loop().time() + timeout
                    remaining = max(0.0, deadline - asyncio.get_event_loop().time())
                    while remaining > 0:
                        try:
                            msg = await asyncio.wait_for(queue.get(), timeout=remaining)
                        except asyncio.TimeoutError:
                            break

                        # stdio stdout lines may contain JSON objects or arrays
                        try:
                            parsed = orjson.loads(msg)
                        except (orjson.JSONDecodeError, ValueError):
                            # not JSON -> skip
                            remaining = max(0.0, deadline - asyncio.get_event_loop().time())
                            continue

                        candidates = parsed if isinstance(parsed, list) else [parsed]
                        for candidate in candidates:
                            if isinstance(candidate, dict) and candidate.get("id") == obj.get("id"):
                                # return the matched response as JSON
                                return ORJSONResponse(candidate)

                        remaining = max(0.0, deadline - asyncio.get_event_loop().time())

                    # timeout -> accept and return 202
                    return PlainTextResponse("accepted (no response yet)", status_code=status.HTTP_202_ACCEPTED)
                finally:
                    if pubsub:
                        pubsub.unsubscribe(queue)

            # Notification -> return 202
            return PlainTextResponse("accepted", status_code=status.HTTP_202_ACCEPTED)

        # ASGI wrapper to route GET/other /mcp scopes to streamable_manager.handle_request
        async def mcp_asgi_wrapper(scope: Scope, receive: Receive, send: Send) -> None:
            """
            ASGI middleware that intercepts HTTP requests to the `/mcp` endpoint.

            If the request is an HTTP call to `/mcp` and a `streamable_manager` is available,
            it can handle the request (currently commented out). All other requests are
            passed to the original FastAPI application.

            Args:
                scope (Scope): The ASGI scope dictionary containing request metadata.
                receive (Receive): An awaitable that yields incoming ASGI events.
                send (Send): An awaitable used to send ASGI events.
            """
            await original_app(scope, receive, send)

        # Replace the app used by uvicorn with the ASGI wrapper
        app = mcp_asgi_wrapper  # type: ignore[assignment]

    # ---------------------- Server lifecycle ----------------------
    config = uvicorn.Config(
        app,
        host=host,
        port=port,
        log_level=log_level,
        lifespan="off",
    )
    server = uvicorn.Server(config)

    shutting_down = asyncio.Event()

    async def _shutdown() -> None:
        """Handle graceful shutdown."""
        if shutting_down.is_set():
            return
        shutting_down.set()
        LOGGER.info("Shutting down multi-protocol server...")
        if stdio:
            await stdio.stop()
        # Streamable HTTP cleanup handled by server shutdown
        # Graceful shutdown by setting the shutdown event
        # Use getattr to safely access should_exit attribute
        setattr(server, "should_exit", getattr(server, "should_exit", False) or True)

    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        with suppress(NotImplementedError):
            loop.add_signal_handler(sig, lambda *_: asyncio.create_task(_shutdown()))

    # If we have a streamable manager, start its context so it can accept ASGI /mcp
    if streamable_manager:
        streamable_context = streamable_manager.run()
        await streamable_context.__aenter__()  # noqa: PLC2801  # pylint: disable=unnecessary-dunder-call,no-member

    # Log available endpoints
    endpoints = []
    if expose_sse:
        endpoints.append(f"SSE: http://{host}:{port}{sse_path}")
    if expose_streamable_http:
        endpoints.append(f"StreamableHTTP: http://{host}:{port}/mcp")

    LOGGER.info(f"Multi-protocol server ready → {', '.join(endpoints)}")

    try:
        await server.serve()
    finally:
        await _shutdown()
        # Clean up streamable HTTP context with timeout to prevent spin loop
        # if tasks don't respond to cancellation (anyio _deliver_cancellation issue)
        if streamable_context:
            # Bounded cleanup window for transport __aexit__. Hardcoded 5s; the
            # previous settings.mcp_session_pool_cleanup_timeout knob disappeared
            # with the pool config in #4205 and no deployment tuned it.
            cleanup_timeout = 5.0
            # Use anyio.move_on_after instead of asyncio.wait_for to properly propagate
            # cancellation through anyio's cancel scope system (prevents orphaned spinning tasks)
            with anyio.move_on_after(cleanup_timeout) as cleanup_scope:
                try:
                    await streamable_context.__aexit__(None, None, None)  # pylint: disable=unnecessary-dunder-call,no-member
                except Exception as e:
                    LOGGER.debug(f"Error cleaning up streamable HTTP context: {e}")
            if cleanup_scope.cancelled_caught:
                LOGGER.warning("Streamable HTTP context cleanup timed out - proceeding anyway")


async def _simple_sse_pump(client: "Any", url: str, max_retries: int, initial_retry_delay: float) -> None:
    """Simple SSE pump that just prints messages to stdout.

    Used when no stdio command is provided to bridge SSE to stdout directly.

    Args:
        client: The HTTP client to use for SSE streaming.
        url: The SSE endpoint URL to connect to.
        max_retries: Maximum number of connection retry attempts.
        initial_retry_delay: Initial delay between retries in seconds.

    Raises:
        HTTPStatusError: If the SSE endpoint returns a non-200 status code.
        Exception: For unexpected errors in SSE stream processing.
    """
    retry_delay = initial_retry_delay
    retry_count = 0

    while retry_count < max_retries:
        try:
            LOGGER.info(f"Connecting to SSE endpoint: {url}")

            async with client.stream("GET", url) as response:
                # Check status code if available (real httpx response)
                if hasattr(response, "status_code") and response.status_code != 200:
                    if httpx:
                        raise httpx.HTTPStatusError(f"SSE endpoint returned {response.status_code}", request=response.request, response=response)
                    raise Exception(f"SSE endpoint returned {response.status_code}")

                # Reset retry counter on successful connection
                retry_count = 0
                retry_delay = initial_retry_delay
                current_event: Optional[SSEEvent] = None

                async for line in response.aiter_lines():
                    event, is_complete = SSEEvent.parse_sse_line(line, current_event)
                    current_event = event

                    if is_complete and current_event:
                        if current_event.event == "endpoint":
                            LOGGER.info(f"Received message endpoint: {current_event.data}")
                        elif current_event.event == "message":
                            # Just print the message to stdout
                            print(current_event.data)
                        elif current_event.event == "keepalive":
                            LOGGER.debug("Received keepalive")

                        # Reset for next event
                        current_event = None

        except Exception as e:
            # Check if it's one of the expected httpx exceptions
            if httpx and isinstance(e, (httpx.ConnectError, httpx.HTTPStatusError, httpx.ReadTimeout)):
                retry_count += 1
                if retry_count >= max_retries:
                    LOGGER.error(f"Max retries ({max_retries}) exceeded. Giving up.")
                    raise

                LOGGER.warning(f"Connection error: {e}. Retrying in {retry_delay}s... (attempt {retry_count}/{max_retries})")
                await asyncio.sleep(retry_delay)
                retry_delay = min(retry_delay * 2, 30)  # Exponential backoff, max 30s
            else:
                LOGGER.error(f"Unexpected error in SSE stream: {e}")
                raise


def start_streamable_http_stdio(
    cmd: str,
    port: int,
    log_level: str,
    cors: Optional[List[str]],
    host: str = "127.0.0.1",
    stateless: bool = False,
    json_response: bool = False,
) -> None:
    """Start stdio to streamable HTTP bridge.

    Entry point for starting a stdio to streamable HTTP bridge server.

    Args:
        cmd: The command to run as a stdio subprocess.
        port: The port to bind the HTTP server to.
        log_level: The logging level to use.
        cors: Optional list of CORS allowed origins.
        host: The host interface to bind to. Defaults to "127.0.0.1".
        stateless: Whether to use stateless mode. Defaults to False.
        json_response: Whether to return JSON responses. Defaults to False.

    Returns:
        None: This function does not return a value.
    """
    return asyncio.run(_run_stdio_to_streamable_http(cmd, port, log_level, cors, host, stateless, json_response))


def start_streamable_http_client(url: str, bearer_token: Optional[str] = None, timeout: float = 30.0, stdio_command: Optional[str] = None) -> None:
    """Start streamable HTTP to stdio bridge.

    Entry point for starting a streamable HTTP to stdio bridge client.

    Args:
        url: The streamable HTTP endpoint URL to connect to.
        bearer_token: Optional OAuth2 bearer token for authentication. Defaults to None.
        timeout: HTTP client timeout in seconds. Defaults to 30.0.
        stdio_command: Optional command to run for local stdio processing.

    Returns:
        None: This function does not return a value.
    """
    return asyncio.run(_run_streamable_http_to_stdio(url, bearer_token, timeout, stdio_command))


def start_stdio(
    cmd: str, port: int, log_level: str, cors: Optional[List[str]], host: str = "127.0.0.1", sse_path: str = "/sse", message_path: str = "/message", keep_alive: float = KEEP_ALIVE_INTERVAL
) -> None:
    """Start stdio bridge.

    Entry point for starting a stdio to SSE bridge server.

    Args:
        cmd: The command to run as a stdio subprocess.
        port: The port to bind the HTTP server to.
        log_level: The logging level to use.
        cors: Optional list of CORS allowed origins.
        host: The host interface to bind to. Defaults to "127.0.0.1".
        sse_path: Path for the SSE endpoint. Defaults to "/sse".
        message_path: Path for the message endpoint. Defaults to "/message".
        keep_alive: Keep-alive interval in seconds. Defaults to KEEP_ALIVE_INTERVAL.

    Returns:
        None: This function does not return a value.

    Examples:
        >>> # Test parameter validation
        >>> isinstance(KEEP_ALIVE_INTERVAL, int)
        True
        >>> KEEP_ALIVE_INTERVAL > 0
        True
        >>> start_stdio("uvx mcp-server-git", 9000, "info", None)  # doctest: +SKIP
    """
    return asyncio.run(_run_stdio_to_sse(cmd, port, log_level, cors, host, sse_path, message_path, keep_alive))


def start_sse(url: str, bearer_token: Optional[str] = None, timeout: float = 30.0, stdio_command: Optional[str] = None) -> None:
    """Start SSE bridge.

    Entry point for starting an SSE to stdio bridge client.

    Examples:
        >>> # Test parameter defaults
        >>> timeout_default = 30.0
        >>> isinstance(timeout_default, float)
        True
        >>> timeout_default > 0
        True

    Args:
        url: The SSE endpoint URL to connect to.
        bearer_token: Optional OAuth2 bearer token for authentication. Defaults to None.
        timeout: HTTP client timeout in seconds. Defaults to 30.0.
        stdio_command: Optional command to run for local stdio processing.

    Returns:
        None: This function does not return a value.

    Examples:
        >>> start_sse("http://example.com/sse", "token123")  # doctest: +SKIP
    """
    return asyncio.run(_run_sse_to_stdio(url, bearer_token, timeout, stdio_command))


def main(argv: Optional[Sequence[str]] | None = None) -> None:
    """Entry point for the translate module.

    Configures logging, parses arguments, and starts the appropriate bridge
    based on command line options. Handles keyboard interrupts gracefully.

    Args:
        argv: Optional sequence of command line arguments. If None, uses sys.argv[1:].

    Examples:
        >>> # Test argument parsing
        >>> try:
        ...     main(["--stdio", "cat", "--port", "9000"])  # doctest: +SKIP
        ... except SystemExit:
        ...     pass  # Would normally start the server
    """
    args = _parse_args(argv or sys.argv[1:])
    logging.basicConfig(
        level=getattr(logging, args.logLevel.upper(), logging.INFO),
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
        datefmt="%Y-%m-%dT%H:%M:%S",
    )

    # Parse header mappings if dynamic environment injection is enabled
    # Pre-normalize mappings once at startup for O(1) lookups per request
    header_mappings: NormalizedMappings | None = None
    if getattr(args, "enable_dynamic_env", False):
        try:
            raw_mappings = parse_header_mappings(getattr(args, "header_to_env", []))
            header_mappings = NormalizedMappings(raw_mappings)
            LOGGER.info(f"Dynamic environment injection enabled with {len(header_mappings)} header mappings")
        except Exception as e:
            LOGGER.error(f"Failed to parse header mappings: {e}")
            raise

    try:
        # Handle gRPC server exposure
        if getattr(args, "grpc", None):
            # First-Party
            from mcpgateway.translate_grpc import expose_grpc_via_sse  # pylint: disable=import-outside-toplevel

            # Parse metadata
            metadata = {}
            if getattr(args, "grpc_metadata", None):
                for item in args.grpc_metadata:
                    if "=" in item:
                        key, value = item.split("=", 1)
                        metadata[key] = value

            asyncio.run(
                expose_grpc_via_sse(
                    target=args.grpc,
                    port=args.port,
                    tls_enabled=getattr(args, "grpc_tls", False),
                    tls_cert=getattr(args, "grpc_cert", None),
                    tls_key=getattr(args, "grpc_key", None),
                    metadata=metadata,
                )
            )

        # Handle local stdio server exposure
        elif args.stdio:
            # Check which protocols to expose
            expose_sse = getattr(args, "expose_sse", False)
            expose_streamable_http = getattr(args, "expose_streamable_http", False)

            # If no protocol specified, default to SSE for backward compatibility
            if not expose_sse and not expose_streamable_http:
                expose_sse = True

            # Use multi-protocol server
            asyncio.run(
                _run_multi_protocol_server(
                    cmd=args.stdio,
                    port=args.port,
                    log_level=args.logLevel,
                    cors=args.cors,
                    host=args.host,
                    expose_sse=expose_sse,
                    expose_streamable_http=expose_streamable_http,
                    sse_path=getattr(args, "ssePath", "/sse"),
                    message_path=getattr(args, "messagePath", "/message"),
                    keep_alive=getattr(args, "keepAlive", KEEP_ALIVE_INTERVAL),
                    stateless=getattr(args, "stateless", False),
                    json_response=getattr(args, "jsonResponse", False),
                    header_mappings=header_mappings,
                )
            )

        # Handle remote connection modes
        elif getattr(args, "connect_sse", None):
            start_sse(args.connect_sse, args.oauth2Bearer, 30.0, args.stdioCommand)
        elif getattr(args, "connect_streamable_http", None):
            start_streamable_http_client(args.connect_streamable_http, args.oauth2Bearer, 30.0, args.stdioCommand)
        elif getattr(args, "connect_grpc", None):
            print("Error: --connect-grpc mode not yet implemented. Use --grpc to expose a gRPC server.", file=sys.stderr)
            sys.exit(1)
        else:
            print("Error: Must specify either --stdio (to expose local server), --grpc (to expose gRPC server), or --connect-sse/--connect-streamable-http (to connect to remote)", file=sys.stderr)
            sys.exit(1)
    except KeyboardInterrupt:
        print("")  # restore shell prompt
        sys.exit(0)
    except (NotImplementedError, ImportError) as exc:
        print(exc, file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":  # python3 -m mcpgateway.translate ...
    main()
