mirror of
https://github.com/agentscope-ai/ReMe.git
synced 2026-09-07 08:26:06 +00:00
3d2ecc60d2
* 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.
249 lines
9.7 KiB
Python
249 lines
9.7 KiB
Python
"""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, 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
|
|
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
|
|
|
|
|
|
# uvicorn 0.41 still imports these deprecated websockets symbols on startup,
|
|
# even though we don't use WebSocket. Silence just those specific warnings.
|
|
_WEBSOCKET_DEPRECATION_PATTERNS = (
|
|
r".*websockets\.legacy is deprecated.*",
|
|
r".*WebSocketServerProtocol is deprecated.*",
|
|
)
|
|
|
|
|
|
@R.register("http")
|
|
class HttpService(BaseService):
|
|
"""Expose jobs through JSON/SSE endpoints and streamable HTTP MCP."""
|
|
|
|
def __init__(
|
|
self,
|
|
host: str = REME_DEFAULT_HOST,
|
|
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)
|
|
self.host: str = host
|
|
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 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=lifespan,
|
|
)
|
|
cors_origins = ["*"]
|
|
self.service.add_middleware(
|
|
CORSMiddleware, # type: ignore[arg-type]
|
|
allow_origins=cors_origins,
|
|
allow_credentials="*" not in cors_origins,
|
|
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:
|
|
"""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,
|
|
)
|
|
uvicorn.run(self.service, host=self.host, port=self.port, **self.kwargs)
|
|
|
|
def finalize_service(self, app: "Application") -> None:
|
|
"""Serve the optional workspace UI after all job endpoints are registered."""
|
|
del app
|
|
if not self.web_enabled:
|
|
return
|
|
|
|
static_dir = resolve_web_static_dir(self.web_static_dir)
|
|
if static_dir is None:
|
|
self.logger.info("Web workspace is unavailable; no static build was found")
|
|
return
|
|
|
|
index_file = static_dir / "index.html"
|
|
assets_dir = static_dir / "assets"
|
|
if assets_dir.is_dir():
|
|
self.service.mount(
|
|
"/assets",
|
|
StaticFiles(directory=str(assets_dir)),
|
|
name="web-assets",
|
|
)
|
|
|
|
no_cache_headers = {"Cache-Control": "no-cache, no-store, must-revalidate"}
|
|
post_only_paths = {
|
|
route.path
|
|
for route in self.service.routes
|
|
if "POST" in (getattr(route, "methods", None) or set())
|
|
and "GET" not in (getattr(route, "methods", None) or set())
|
|
}
|
|
|
|
@self.service.get("/{full_path:path}", include_in_schema=False)
|
|
async def workspace_spa(full_path: str):
|
|
if full_path in {"docs", "redoc", "openapi.json"}:
|
|
raise HTTPException(status_code=404, detail="Not Found")
|
|
if f"/{full_path}" in post_only_paths:
|
|
raise HTTPException(
|
|
status_code=405,
|
|
detail="Method Not Allowed",
|
|
headers={"Allow": "POST"},
|
|
)
|
|
|
|
if full_path and not Path(full_path).is_absolute():
|
|
static_file = (static_dir / full_path).resolve()
|
|
if static_file.is_relative_to(static_dir) and static_file.is_file():
|
|
return FileResponse(static_file)
|
|
|
|
return FileResponse(index_file, headers=no_cache_headers)
|
|
|
|
# ----- 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."""
|
|
|
|
async def endpoint(request: Request) -> Response:
|
|
return await job(**request.model_dump(exclude_none=True))
|
|
|
|
self.service.post(
|
|
f"/{job.name}",
|
|
response_model=Response,
|
|
description=job.description,
|
|
)(endpoint)
|
|
|
|
def _add_stream_job(self, job: StreamJob) -> None:
|
|
"""Register a StreamJob as POST /{job.name} streaming chunks as text/event-stream."""
|
|
|
|
async def endpoint(request: Request) -> StreamingResponse:
|
|
stream_queue: asyncio.Queue = asyncio.Queue()
|
|
task = asyncio.create_task(
|
|
job(stream_queue=stream_queue, **request.model_dump(exclude_none=True)),
|
|
)
|
|
|
|
async def body() -> AsyncGenerator[bytes, None]:
|
|
async for chunk in execute_stream_task(
|
|
stream_queue=stream_queue,
|
|
task=task,
|
|
task_name=job.name,
|
|
output_format="bytes",
|
|
):
|
|
assert isinstance(chunk, bytes)
|
|
yield chunk
|
|
|
|
return StreamingResponse(body(), media_type="text/event-stream")
|
|
|
|
self.service.post(f"/{job.name}")(endpoint)
|