From 332ac780b7212867bd21ea9a3e586abd24e5df66 Mon Sep 17 00:00:00 2001
From: "jinli.yl" <jinli.yl@alibaba-inc.com>
Date: Tue, 28 Apr 2026 00:23:51 +0800
Subject: [PATCH] feat(mcp): add Model Context Protocol service implementation

- Create MCPService class to expose jobs as MCP tools
- Implement FastMCP integration with async lifespan management
- Add environment variable setup for service information
- Implement tool registration for job execution via FunctionTool
- Support multiple transport types (sse, stdio) with configurable host/port
- Integrate with application lifecycle for proper startup/shutdown
---
 reme2/component/service/mcp_service.py | 71 ++++++++++++++++++++++++++
 1 file changed, 71 insertions(+)
 create mode 100644 reme2/component/service/mcp_service.py

diff --git a/reme2/component/service/mcp_service.py b/reme2/component/service/mcp_service.py
new file mode 100644
index 00000000..c662dee5
--- /dev/null
+++ b/reme2/component/service/mcp_service.py
@@ -0,0 +1,71 @@
+"""Model Context Protocol (MCP) service implementation."""
+import json
+import os
+from contextlib import asynccontextmanager
+from typing import TYPE_CHECKING
+
+from fastmcp import FastMCP
+from fastmcp.server.server import Transport
+from fastmcp.tools import FunctionTool
+
+from .base_service import BaseService
+from ..component_registry import R
+from ..job import StreamJob
+from ...constants import REME_DEFAULT_HOST, REME_DEFAULT_PORT, REME_SERVICE_INFO
+
+if TYPE_CHECKING:
+    from ...application import Application
+    from ..job import BaseJob
+
+
+@R.register("mcp")
+class MCPService(BaseService):
+    """Expose jobs as Model Context Protocol (MCP) tools."""
+
+    def __init__(
+            self,
+            transport: Transport = "sse",
+            host: str = REME_DEFAULT_HOST,
+            port: int = REME_DEFAULT_PORT,
+            **kwargs):
+        super().__init__(**kwargs)
+        self.transport: Transport = transport
+        self.host: str = host
+        self.port: int = port
+
+    def build_service(self, app: "Application") -> None:
+
+        @asynccontextmanager
+        async def lifespan(_: FastMCP):
+            await app.start()
+            service_info = json.dumps({"host": self.host, "port": self.port})
+            os.environ[REME_SERVICE_INFO] = service_info
+            self.logger.info(f"ReMe MCP Service started: {REME_SERVICE_INFO}={service_info}")
+            yield
+            await app.close()
+
+        self.service = FastMCP(name=app.config.app_name, lifespan=lifespan)
+
+    def add_job(self, job: "BaseJob") -> None:
+        if isinstance(job, StreamJob):
+            return
+
+        async def execute_tool(**kwargs):
+            response = await job(**kwargs)
+            return response.answer
+
+        self.service.add_tool(
+            FunctionTool(
+                name=job.name,
+                description=job.description,
+                fn=execute_tool,
+                parameters=job.parameters if job.parameters else None,
+            ),
+        )
+
+    def start_service(self, app: "Application") -> None:
+        transport_kwargs = {}
+        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)