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: