feat(service): expose MCP through HTTP backend (#498)

* feat(service): expose MCP through HTTP backend

Serve JSON/SSE job endpoints and streamable HTTP MCP from one FastAPI application, sharing the same jobs and application lifecycle. Preserve the standalone MCP backend, add configurable MCP HTTP settings, update startup metadata and integration docs, and cover routing, lifecycle, configuration, and compatibility behavior with unit tests.

* fix(service): preserve MCP request protections

Route the exact MCP path through the complete FastMCP ASGI application so its middleware and state remain active. Reject non-literal MCP paths and validate reserved Job conflicts before tolerant service registration. Add regression coverage for middleware preservation, route syntax, and startup failure.

* fix(service): reject encoded MCP paths

Reject percent signs in mcp_path so ASGI path decoding cannot turn an accepted configuration into an unreachable route. Cover encoded slash, space, and double-encoded slash inputs.

integrations/claude_code/README.md

@@ -40,13 +40,13 @@ server means one set of background watchers / dream cron across all your Claude
    LLM_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
    ```
 
-3. Start the ReMe MCP server (one time, leave it running):
+3. Start the ReMe HTTP server (one time, leave it running):
 
    ```bash
-   reme start service.backend=mcp service.transport=streamable-http
+   reme start service.backend=http
    ```
 
-   It serves `http://127.0.0.1:2333/mcp`. To use a different port, start with
+   The same process serves the JSON Job API and MCP at `http://127.0.0.1:2333/mcp`. To use a different port, start with
    `service.port=<port>` and update the `url` in `.mcp.json` to match.
 
 ## Install the plugin

integrations/claude_code/reme/skills/reme-memory/SKILL.md

@@ -14,7 +14,7 @@ The recall tools come from the `reme` MCP server (surfaced as `mcp__reme__…`):
 running:
 
 ```
-reme start service.backend=mcp service.transport=streamable-http
+reme start service.backend=http
 ```
 
 If the tools are missing, that server is not running — tell the user the command above instead of

plugins/auto-fin/README.md

@@ -49,14 +49,14 @@ curl -s http://127.0.0.1:2333/auto_fin \
   -d '{"topics":"黄金,AI,存储芯片"}'
 ```
 
-When enabled on an MCP service, the same Job is exposed as the `auto_fin` MCP tool. The default topics are
+The HTTP service also exposes the same Job as the `auto_fin` MCP tool at `/mcp`. The default topics are
 `黄金,机器人,半导体`; an empty value also uses these defaults.
 
-To host the same application as an MCP service instead:
+To host the application with both JSON and MCP access:
 
 ```bash
 reme start plugins='["auto-fin"]' \
-  service.backend=mcp service.transport=streamable-http
+  service.backend=http
 ```
 
 To add Auto Fin to another application instead, select that config explicitly, for example:

plugins/auto-fin/README_ZH.md

@@ -44,14 +44,14 @@ curl -s http://127.0.0.1:2333/auto_fin \
   -d '{"topics":"黄金,AI,存储芯片"}'
 ```
 
-在 MCP service 中启用插件时，同一个 Job 会暴露为 `auto_fin` MCP tool。默认 topics 是 `黄金,机器人,半导体`，
+HTTP service 也会在 `/mcp` 中将同一个 Job 暴露为 `auto_fin` MCP tool。默认 topics 是 `黄金,机器人,半导体`，
 传入空值也会使用默认值。
 
-如果需要将同一个应用作为 MCP service 启动：
+如果需要同时通过 JSON 和 MCP 访问同一个应用：
 
 ```bash
 reme start plugins='["auto-fin"]' \
-  service.backend=mcp service.transport=streamable-http
+  service.backend=http
 ```
 
 如果需要将 Auto Fin 叠加到其他应用，则显式选择相应配置，例如：

reme/components/service/http_service.py

@@ -1,16 +1,17 @@
-"""HTTP service: exposes jobs as FastAPI endpoints (JSON, or SSE for stream jobs)."""
+"""HTTP service: expose jobs through JSON/SSE endpoints and MCP tools."""
 
 import asyncio
 import warnings
 from collections.abc import AsyncGenerator
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 import uvicorn
 from fastapi import FastAPI, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import FileResponse, StreamingResponse
 from fastapi.staticfiles import StaticFiles
+from starlette.routing import Route
 
 from .base_service import BaseService
 from ..component_registry import R
@@ -18,6 +19,7 @@ from ..job import BaseJob, StreamJob
 from ...constants import REME_DEFAULT_HOST, REME_DEFAULT_PORT
 from ...schema import Request, Response
 from ...utils import execute_stream_task, resolve_web_static_dir
+from .mcp_tools import add_mcp_job
 
 if TYPE_CHECKING:
     from ...application import Application
@@ -33,7 +35,7 @@ _WEBSOCKET_DEPRECATION_PATTERNS = (
 
 @R.register("http")
 class HttpService(BaseService):
-    """Map non-stream jobs to JSON POST endpoints and StreamJobs to SSE endpoints."""
+    """Expose jobs through JSON/SSE endpoints and streamable HTTP MCP."""
 
     def __init__(
         self,
@@ -41,6 +43,11 @@ class HttpService(BaseService):
         port: int = REME_DEFAULT_PORT,
         web_enabled: bool = True,
         web_static_dir: str | None = None,
+        mcp_enabled: bool = True,
+        mcp_path: str = "/mcp",
+        mcp_stateless_http: bool = False,
+        injected_job_kwargs: dict[str, Any] | None = None,
+        tool_error_on_failure: bool = False,
         **kwargs,
     ):
         super().__init__(**kwargs)
@@ -48,14 +55,34 @@ class HttpService(BaseService):
         self.port: int = port
         self.web_enabled = web_enabled
         self.web_static_dir = web_static_dir
+        self.mcp_enabled = mcp_enabled
+        self.mcp_path = self._validate_mcp_path(mcp_path)
+        self.mcp_stateless_http = mcp_stateless_http
+        self.injected_job_kwargs = dict(injected_job_kwargs or {})
+        self.tool_error_on_failure = tool_error_on_failure
+        self.mcp_server = None
+        self.mcp_app = None
 
     # ----- BaseService contract ------------------------------------------
 
     def build_service(self, app: "Application") -> None:
-        """Create the FastAPI app with permissive CORS and an app-managed lifespan."""
+        """Create one FastAPI app containing JSON/SSE and optional MCP routes."""
+        lifespan = self._lifespan(app, self.host, self.port)
+        if self.mcp_enabled:
+            from fastmcp import FastMCP
+            from fastmcp.utilities.lifespan import combine_lifespans
+
+            self.mcp_server = FastMCP(name=app.config.app_name)
+            self.mcp_app = self.mcp_server.http_app(
+                path=self.mcp_path,
+                transport="streamable-http",
+                stateless_http=self.mcp_stateless_http,
+            )
+            lifespan = combine_lifespans(lifespan, self.mcp_app.lifespan)
+
         self.service = FastAPI(
             title=app.config.app_name,
-            lifespan=self._lifespan(app, self.host, self.port),
+            lifespan=lifespan,
         )
         cors_origins = ["*"]
         self.service.add_middleware(
@@ -65,19 +92,60 @@ class HttpService(BaseService):
             allow_methods=["*"],
             allow_headers=["*"],
         )
+        if self.mcp_app is not None:
+            # Forward the exact path to the complete FastMCP ASGI app. Copying
+            # only its routes would bypass its middleware and application state;
+            # mounting it would make the trailing-slash path canonical instead.
+            self.service.router.routes.append(
+                Route(
+                    self.mcp_path,
+                    endpoint=self.mcp_app,
+                    include_in_schema=False,
+                ),
+            )
+
+    def add_jobs(self, app: "Application") -> None:
+        """Validate reserved routes before the shared tolerant registration loop."""
+        if self.mcp_enabled:
+            conflicts = sorted(
+                job.name
+                for name, job in app.context.jobs.items()
+                if job.enable_serve and (self.jobs is None or name in self.jobs) and f"/{job.name}" == self.mcp_path
+            )
+            if conflicts:
+                names = ", ".join(conflicts)
+                raise ValueError(
+                    f"Job name conflicts with the MCP endpoint {self.mcp_path!r}: {names}",
+                )
+        super().add_jobs(app)
 
     def add_job(self, job: BaseJob) -> bool:
-        """Dispatch to streaming or non-streaming registration based on job type."""
+        """Register HTTP routes for every job and MCP tools for non-stream jobs."""
+        if self.mcp_enabled and f"/{job.name}" == self.mcp_path:
+            raise ValueError(
+                f"Job name '{job.name}' conflicts with the MCP endpoint {self.mcp_path!r}",
+            )
         if isinstance(job, StreamJob):
             self._add_stream_job(job)
         else:
             self._add_json_job(job)
+            if self.mcp_server is not None:
+                add_mcp_job(
+                    self.mcp_server,
+                    job,
+                    injected_job_kwargs=self.injected_job_kwargs,
+                    tool_error_on_failure=self.tool_error_on_failure,
+                )
         return True
 
     def start_service(self, app: "Application") -> None:
         """Run uvicorn, suppressing unrelated websocket deprecation noise."""
         for pattern in _WEBSOCKET_DEPRECATION_PATTERNS:
-            warnings.filterwarnings("ignore", category=DeprecationWarning, message=pattern)
+            warnings.filterwarnings(
+                "ignore",
+                category=DeprecationWarning,
+                message=pattern,
+            )
         uvicorn.run(self.service, host=self.host, port=self.port, **self.kwargs)
 
     def finalize_service(self, app: "Application") -> None:
@@ -128,6 +196,23 @@ class HttpService(BaseService):
 
     # ----- Endpoint factories --------------------------------------------
 
+    @staticmethod
+    def _validate_mcp_path(path: str) -> str:
+        """Return a canonical, non-reserved absolute path for the MCP endpoint."""
+        if not path.startswith("/") or path == "/" or path.endswith("/"):
+            raise ValueError(
+                "mcp_path must start with '/', must not be '/', and must not end with '/'",
+            )
+        if "//" in path or any(segment in {".", ".."} for segment in path.split("/")):
+            raise ValueError("mcp_path must use non-empty literal path segments")
+        if any(char in path for char in "{}?#%\\") or any(
+            char.isspace() or ord(char) < 32 or ord(char) == 127 for char in path
+        ):
+            raise ValueError("mcp_path must be a literal URL path without route, query, or fragment syntax")
+        if path in {"/assets", "/docs", "/redoc", "/openapi.json"}:
+            raise ValueError(f"mcp_path conflicts with reserved HTTP path {path!r}")
+        return path
+
     def _add_json_job(self, job: BaseJob) -> None:
         """Register a job as POST /{job.name} returning a JSON Response."""
 

reme/components/service/mcp_service.py

@@ -4,8 +4,9 @@ from typing import TYPE_CHECKING, Any
 
 from .base_service import BaseService
 from ..component_registry import R
-from ..job import BaseJob, StreamJob
+from ..job import BaseJob
 from ...constants import REME_DEFAULT_HOST, REME_DEFAULT_PORT
+from .mcp_tools import add_mcp_job
 
 if TYPE_CHECKING:
     from fastmcp.server.server import Transport
@@ -45,41 +46,12 @@ class MCPService(BaseService):
 
     def add_job(self, job: BaseJob) -> bool:
         """Register a non-stream job as an MCP tool; StreamJobs are unsupported."""
-        from fastmcp.exceptions import ToolError
-        from fastmcp.tools import FunctionTool
-
-        if isinstance(job, StreamJob):
-            return False
-
-        async def execute_tool(**kwargs):
-            conflicts = sorted(self.injected_job_kwargs.keys() & kwargs.keys())
-            if conflicts:
-                names = ", ".join(conflicts)
-                raise ToolError(f"{names} injected by the MCP server and cannot be provided by the caller")
-            kwargs.update(self.injected_job_kwargs)
-            response = await job(**kwargs)
-            if self.tool_error_on_failure and not response.success:
-                raise ToolError(str(response.answer))
-            return response.answer
-
-        parameters = dict(job.parameters or {})
-        injected_names = self.injected_job_kwargs.keys()
-        if "properties" in parameters:
-            parameters["properties"] = {
-                name: schema for name, schema in parameters["properties"].items() if name not in injected_names
-            }
-        if "required" in parameters:
-            parameters["required"] = [name for name in parameters["required"] if name not in injected_names]
-
-        self.service.add_tool(
-            FunctionTool(
-                name=job.name,
-                description=job.description,
-                fn=execute_tool,
-                parameters=parameters,
-            ),
+        return add_mcp_job(
+            self.service,
+            job,
+            injected_job_kwargs=self.injected_job_kwargs,
+            tool_error_on_failure=self.tool_error_on_failure,
         )
-        return True
 
     def start_service(self, app: "Application") -> None:
         """Run the MCP server; bind host/port only for network transports."""
@@ -87,4 +59,8 @@ class MCPService(BaseService):
         if self.transport != "stdio":
             transport_kwargs["host"] = self.host
             transport_kwargs["port"] = self.port
-        self.service.run(transport=self.transport, show_banner=False, **transport_kwargs)
+        self.service.run(
+            transport=self.transport,
+            show_banner=False,
+            **transport_kwargs,
+        )

reme/components/service/mcp_tools.py

@@ -0,0 +1,52 @@
+"""Shared MCP tool registration for services that expose ReMe jobs."""
+
+from typing import Any
+
+from ..job import BaseJob, StreamJob
+
+
+def add_mcp_job(
+    server: Any,
+    job: BaseJob,
+    *,
+    injected_job_kwargs: dict[str, Any],
+    tool_error_on_failure: bool,
+) -> bool:
+    """Register a non-stream job as an MCP tool on ``server``."""
+    from fastmcp.exceptions import ToolError
+    from fastmcp.tools import FunctionTool
+
+    if isinstance(job, StreamJob):
+        return False
+
+    async def execute_tool(**kwargs):
+        conflicts = sorted(injected_job_kwargs.keys() & kwargs.keys())
+        if conflicts:
+            names = ", ".join(conflicts)
+            raise ToolError(
+                f"{names} injected by the MCP server and cannot be provided by the caller",
+            )
+        kwargs.update(injected_job_kwargs)
+        response = await job(**kwargs)
+        if tool_error_on_failure and not response.success:
+            raise ToolError(str(response.answer))
+        return response.answer
+
+    parameters = dict(job.parameters or {})
+    injected_names = injected_job_kwargs.keys()
+    if "properties" in parameters:
+        parameters["properties"] = {
+            name: schema for name, schema in parameters["properties"].items() if name not in injected_names
+        }
+    if "required" in parameters:
+        parameters["required"] = [name for name in parameters["required"] if name not in injected_names]
+
+    server.add_tool(
+        FunctionTool(
+            name=job.name,
+            description=job.description,
+            fn=execute_tool,
+            parameters=parameters,
+        ),
+    )
+    return True

reme/config/default.yaml

@@ -1,6 +1,8 @@
 service:
   backend: http
   web_enabled: true
+  mcp_enabled: true
+  mcp_path: /mcp
 
 jobs:
   index_update_loop:

reme/utils/logo_utils.py

@@ -79,7 +79,13 @@ def print_logo(app_config: "ApplicationConfig", runtime_service: "BaseService |
             host = getattr(runtime_service, "host", extra.get("host", REME_DEFAULT_HOST))
             port = getattr(runtime_service, "port", extra.get("port", REME_DEFAULT_PORT))
             info_table.add_row("🔗", "URL:", f"http://{host}:{port}")
+            mcp_enabled = getattr(runtime_service, "mcp_enabled", extra.get("mcp_enabled", True))
+            if mcp_enabled:
+                mcp_path = getattr(runtime_service, "mcp_path", extra.get("mcp_path", "/mcp"))
+                info_table.add_row("🚌", "MCP:", f"http://{host}:{port}{mcp_path}")
             info_table.add_row("📚", "FastAPI:", Text(get_version("fastapi"), style="dim"))
+            if mcp_enabled:
+                info_table.add_row("📚", "FastMCP:", Text(get_version("fastmcp"), style="dim"))
         case "mcp":
             transport = getattr(runtime_service, "transport", extra.get("transport", "sse"))
             info_table.add_row("🚌", "Transport:", transport)

tests/unit/test_config_parser.py

@@ -19,6 +19,8 @@ def test_load_builtin_config_by_filename_with_suffix():
     cfg = _load_config("default.yaml")
 
     assert cfg["service"]["backend"] == "http"
+    assert cfg["service"]["mcp_enabled"] is True
+    assert cfg["service"]["mcp_path"] == "/mcp"
 
 
 @pytest.mark.parametrize("provider_count", [1, 2])

tests/unit/test_http_web_workspace.py

@@ -1,13 +1,16 @@
-"""HTTP service coverage for the optional bundled web workspace."""
+"""HTTP service coverage for MCP and the optional bundled web workspace."""
 
+import asyncio
 import sys
 from pathlib import Path
 from types import ModuleType, SimpleNamespace
 
 import pytest
 from fastapi.testclient import TestClient
+from starlette.middleware.base import BaseHTTPMiddleware
 
 from reme.components.service.http_service import HttpService
+from reme.components.job import BaseJob, StreamJob
 from reme.utils import REME_WEB_STATIC_DIR, resolve_web_static_dir
 
 
@@ -30,7 +33,10 @@ def _static_build(tmp_path: Path) -> Path:
     static_dir = tmp_path / "web"
     assets_dir = static_dir / "assets"
     assets_dir.mkdir(parents=True)
-    (static_dir / "index.html").write_text("<main>ReMe workspace</main>", encoding="utf-8")
+    (static_dir / "index.html").write_text(
+        "<main>ReMe workspace</main>",
+        encoding="utf-8",
+    )
     (static_dir / "favicon.svg").write_text("<svg></svg>", encoding="utf-8")
     (assets_dir / "app.js").write_text("console.log('reme')", encoding="utf-8")
     return static_dir
@@ -66,7 +72,10 @@ def test_http_service_serves_workspace_without_shadowing_jobs(tmp_path: Path) ->
 def test_http_service_can_disable_workspace(tmp_path: Path) -> None:
     """Leave the root route unregistered when workspace serving is disabled."""
     app = _FakeApplication()
-    service = HttpService(web_enabled=False, web_static_dir=str(_static_build(tmp_path)))
+    service = HttpService(
+        web_enabled=False,
+        web_static_dir=str(_static_build(tmp_path)),
+    )
     service.build_service(app)  # type: ignore[arg-type]
     service.finalize_service(app)  # type: ignore[arg-type]
 
@@ -74,7 +83,148 @@ def test_http_service_can_disable_workspace(tmp_path: Path) -> None:
         assert client.get("/").status_code == 404
 
 
-def test_http_service_does_not_serve_symlinks_outside_static_dir(tmp_path: Path) -> None:
+def test_http_service_exposes_non_stream_jobs_as_mcp_tools() -> None:
+    """The HTTP backend exposes the same non-stream Job instance through MCP."""
+
+    async def run() -> None:
+        service = HttpService(web_enabled=False)
+        service.build_service(_FakeApplication())  # type: ignore[arg-type]
+        job = BaseJob(name="search", description="Search memories")
+
+        assert service.add_job(job) is True
+        assert service.mcp_server is not None
+        assert await service.mcp_server.get_tool("search") is not None
+        assert any(route.path == "/search" for route in service.service.routes)
+
+    asyncio.run(run())
+
+
+def test_http_service_skips_stream_jobs_for_mcp() -> None:
+    """Stream jobs remain available over HTTP SSE without becoming MCP tools."""
+
+    async def run() -> None:
+        service = HttpService(web_enabled=False)
+        service.build_service(_FakeApplication())  # type: ignore[arg-type]
+
+        assert service.add_job(StreamJob(name="stream")) is True
+        assert service.mcp_server is not None
+        assert await service.mcp_server.get_tool("stream") is None
+        assert any(route.path == "/stream" for route in service.service.routes)
+
+    asyncio.run(run())
+
+
+def test_http_service_uses_exact_mcp_path_and_runs_one_application_lifespan() -> None:
+    """Serve MCP at /mcp while starting and closing the shared Application once."""
+
+    class CountingApplication(_FakeApplication):
+        """Record how often the shared Application lifecycle is entered."""
+
+        def __init__(self) -> None:
+            super().__init__()
+            self.start_count = 0
+            self.close_count = 0
+
+        async def start(self) -> None:
+            self.start_count += 1
+
+        async def close(self) -> None:
+            self.close_count += 1
+
+    app = CountingApplication()
+    service = HttpService(web_enabled=False)
+    service.build_service(app)  # type: ignore[arg-type]
+    mcp_server = service.mcp_server
+
+    class VerifyFastMCPAppMiddleware(BaseHTTPMiddleware):
+        """Prove requests retain FastMCP middleware and application state."""
+
+        async def dispatch(self, request, call_next):
+            """Verify the child app context and mark its response."""
+            assert request.app.state.fastmcp_server is mcp_server
+            response = await call_next(request)
+            response.headers["X-FastMCP-Middleware"] = "preserved"
+            return response
+
+    service.mcp_app.add_middleware(VerifyFastMCPAppMiddleware)
+
+    with TestClient(service.service, follow_redirects=False) as client:
+        response = client.post(
+            "/mcp",
+            headers={
+                "Accept": "application/json, text/event-stream",
+                "Content-Type": "application/json",
+            },
+            json={
+                "jsonrpc": "2.0",
+                "id": 1,
+                "method": "initialize",
+                "params": {
+                    "protocolVersion": "2025-06-18",
+                    "capabilities": {},
+                    "clientInfo": {"name": "test", "version": "1"},
+                },
+            },
+        )
+        assert response.status_code == 200
+        assert '"serverInfo"' in response.text
+        assert response.headers["X-FastMCP-Middleware"] == "preserved"
+        assert client.post("/mcp/mcp", json={}).status_code == 404
+
+    assert app.start_count == 1
+    assert app.close_count == 1
+
+
+def test_http_service_can_disable_mcp() -> None:
+    """Allow deployments to retain the legacy HTTP-only surface explicitly."""
+    service = HttpService(web_enabled=False, mcp_enabled=False)
+    service.build_service(_FakeApplication())  # type: ignore[arg-type]
+
+    with TestClient(service.service) as client:
+        assert client.post("/mcp", json={}).status_code == 404
+
+
+def test_http_service_rejects_invalid_or_conflicting_mcp_paths() -> None:
+    """Reject paths that shadow built-ins and jobs that shadow the MCP endpoint."""
+    for path in (
+        "mcp",
+        "/",
+        "/mcp/",
+        "/docs",
+        "/{rest:path}",
+        "/mcp?mode=test",
+        "/mcp#fragment",
+        "/mcp%2Fv2",
+        "/mcp%20v2",
+        "/mcp%252Fv2",
+        "/mcp path",
+        "/mcp//nested",
+        "/mcp/../nested",
+    ):
+        with pytest.raises(ValueError, match="mcp_path"):
+            HttpService(mcp_path=path)
+
+    service = HttpService(web_enabled=False)
+    service.build_service(_FakeApplication())  # type: ignore[arg-type]
+    with pytest.raises(ValueError, match="conflicts with the MCP endpoint"):
+        service.add_job(BaseJob(name="mcp"))
+
+
+def test_http_service_fails_startup_preflight_for_mcp_job_conflict() -> None:
+    """Do not let BaseService's tolerant registration hide reserved-route conflicts."""
+    service = HttpService(web_enabled=False)
+    service.build_service(_FakeApplication())  # type: ignore[arg-type]
+    app = SimpleNamespace(
+        context=SimpleNamespace(jobs={"mcp": BaseJob(name="mcp")}),
+    )
+
+    with pytest.raises(ValueError, match="conflicts with the MCP endpoint"):
+        service.add_jobs(app)
+
+
+def test_http_service_does_not_serve_symlinks_outside_static_dir(
+    tmp_path: Path,
+) -> None:
     """Do not expose files reached through symlinks outside the static build."""
     static_dir = _static_build(tmp_path)
     secret_file = tmp_path / "secret.txt"
@@ -93,7 +243,10 @@ def test_http_service_does_not_serve_symlinks_outside_static_dir(tmp_path: Path)
         assert client.get("/escape.txt").text == "<main>ReMe workspace</main>"
 
 
-def test_static_dir_configuration_precedes_environment(monkeypatch, tmp_path: Path) -> None:
+def test_static_dir_configuration_precedes_environment(
+    monkeypatch,
+    tmp_path: Path,
+) -> None:
     """Prefer an explicit static directory over the environment setting."""
     configured = _static_build(tmp_path / "configured")
     environment = _static_build(tmp_path / "environment")

tests/unit/test_logo_utils.py

@@ -30,6 +30,7 @@ def test_logo_uses_runtime_http_address(monkeypatch) -> None:
     output = _render_logo(monkeypatch, config, runtime_service)
 
     assert "http://0.0.0.0:8123" in output
+    assert "http://0.0.0.0:8123/mcp" in output
 
 
 def test_logo_fallback_matches_service_defaults(monkeypatch) -> None:
@@ -39,6 +40,18 @@ def test_logo_fallback_matches_service_defaults(monkeypatch) -> None:
     output = _render_logo(monkeypatch, config)
 
     assert f"http://{REME_DEFAULT_HOST}:{REME_DEFAULT_PORT}" in output
+    assert f"http://{REME_DEFAULT_HOST}:{REME_DEFAULT_PORT}/mcp" in output
+
+
+def test_logo_hides_disabled_http_mcp_endpoint(monkeypatch) -> None:
+    """Do not advertise MCP when it is explicitly disabled on the HTTP service."""
+    config = ApplicationConfig(
+        service=ComponentConfig(backend="http", mcp_enabled=False),
+    )
+
+    output = _render_logo(monkeypatch, config)
+
+    assert f"http://{REME_DEFAULT_HOST}:{REME_DEFAULT_PORT}/mcp" not in output
 
 
 def test_logo_uses_runtime_mcp_transport_and_address(monkeypatch) -> None: