From bb60045070f7fb85babbda51c7b68ca85ffe07e2 Mon Sep 17 00:00:00 2001 From: "jinli.yl" Date: Thu, 27 Aug 2026 16:04:42 +0800 Subject: [PATCH] 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. --- integrations/claude_code/README.md | 6 +- .../reme/skills/reme-memory/SKILL.md | 2 +- plugins/auto-fin/README.md | 6 +- plugins/auto-fin/README_ZH.md | 6 +- reme/components/service/http_service.py | 70 +++++++++- reme/components/service/mcp_service.py | 48 ++----- reme/components/service/mcp_tools.py | 52 ++++++++ reme/config/default.yaml | 2 + reme/utils/logo_utils.py | 6 + tests/unit/test_config_parser.py | 2 + tests/unit/test_http_web_workspace.py | 122 +++++++++++++++++- tests/unit/test_logo_utils.py | 13 ++ 12 files changed, 277 insertions(+), 58 deletions(-) create mode 100644 reme/components/service/mcp_tools.py diff --git a/integrations/claude_code/README.md b/integrations/claude_code/README.md index 1693a76b..9bb81969 100644 --- a/integrations/claude_code/README.md +++ b/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=` and update the `url` in `.mcp.json` to match. ## Install the plugin diff --git a/integrations/claude_code/reme/skills/reme-memory/SKILL.md b/integrations/claude_code/reme/skills/reme-memory/SKILL.md index a4672595..c87a1863 100644 --- a/integrations/claude_code/reme/skills/reme-memory/SKILL.md +++ b/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 diff --git a/plugins/auto-fin/README.md b/plugins/auto-fin/README.md index 4ed30329..5a0c2db4 100644 --- a/plugins/auto-fin/README.md +++ b/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: diff --git a/plugins/auto-fin/README_ZH.md b/plugins/auto-fin/README_ZH.md index 2086d076..6dc0a3cf 100644 --- a/plugins/auto-fin/README_ZH.md +++ b/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 叠加到其他应用，则显式选择相应配置，例如： diff --git a/reme/components/service/http_service.py b/reme/components/service/http_service.py index a61369c8..177fec33 100644 --- a/reme/components/service/http_service.py +++ b/reme/components/service/http_service.py @@ -1,10 +1,10 @@ -"""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 @@ -18,6 +18,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 +34,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 +42,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 +54,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 +91,38 @@ class HttpService(BaseService): allow_methods=["*"], allow_headers=["*"], ) + if self.mcp_app is not None: + # Mounting at /mcp makes Starlette redirect to /mcp/. Merge the + # generated routes so the configured path remains canonical. + self.service.router.routes.extend(self.mcp_app.routes) 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 +173,17 @@ 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 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.""" diff --git a/reme/components/service/mcp_service.py b/reme/components/service/mcp_service.py index 4de5b94d..cbab4674 100644 --- a/reme/components/service/mcp_service.py +++ b/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, + ) diff --git a/reme/components/service/mcp_tools.py b/reme/components/service/mcp_tools.py new file mode 100644 index 00000000..6e0b4caf --- /dev/null +++ b/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 diff --git a/reme/config/default.yaml b/reme/config/default.yaml index 95a0bad5..4faa9b60 100644 --- a/reme/config/default.yaml +++ b/reme/config/default.yaml @@ -1,6 +1,8 @@ service: backend: http web_enabled: true + mcp_enabled: true + mcp_path: /mcp jobs: index_update_loop: diff --git a/reme/utils/logo_utils.py b/reme/utils/logo_utils.py index ef016aba..f7eaee5a 100644 --- a/reme/utils/logo_utils.py +++ b/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) diff --git a/tests/unit/test_config_parser.py b/tests/unit/test_config_parser.py index 22c51434..b518db20 100644 --- a/tests/unit/test_config_parser.py +++ b/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]) diff --git a/tests/unit/test_http_web_workspace.py b/tests/unit/test_http_web_workspace.py index 709214e4..57a5a7f8 100644 --- a/tests/unit/test_http_web_workspace.py +++ b/tests/unit/test_http_web_workspace.py @@ -1,5 +1,6 @@ -"""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 @@ -8,6 +9,7 @@ import pytest from fastapi.testclient import TestClient 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 +32,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("
ReMe workspace
", encoding="utf-8") + (static_dir / "index.html").write_text( + "
ReMe workspace
", + encoding="utf-8", + ) (static_dir / "favicon.svg").write_text("", encoding="utf-8") (assets_dir / "app.js").write_text("console.log('reme')", encoding="utf-8") return static_dir @@ -66,7 +71,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 +82,108 @@ 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] + + 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 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"): + 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_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 +202,10 @@ def test_http_service_does_not_serve_symlinks_outside_static_dir(tmp_path: Path) assert client.get("/escape.txt").text == "
ReMe workspace
" -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") diff --git a/tests/unit/test_logo_utils.py b/tests/unit/test_logo_utils.py index 23757ee5..c1ac8e53 100644 --- a/tests/unit/test_logo_utils.py +++ b/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: