diff --git a/backend/pyproject.toml b/backend/pyproject.toml
index 029fda9d..725205a7 100644
--- a/backend/pyproject.toml
+++ b/backend/pyproject.toml
@@ -19,10 +19,11 @@ dependencies = [
     "openai>=1.54.3",
     "semver>=3.1.0",
     "webauthn>=2.8.0",
-    "mcp>=1.8.0",
+    "mcp>=2.3.0",
     "anthropic>=0.116.0",
     "google-genai>=2.10.0",
     "packaging>=26.3",
+    "jsonschema>=4.20.0",
 ]
 
 [dependency-groups]
@@ -82,6 +83,11 @@ ignore_missing_imports = true
 module = "aiohttp"
 ignore_missing_imports = true
 
+# jsonschema 不带类型信息（无 py.typed）
+[[tool.mypy.overrides]]
+module = "jsonschema"
+ignore_missing_imports = true
+
 # 测试代码惯用 `obj.method = AsyncMock()` 直接打桩实例方法，这是标准、安全的
 # unittest.mock 用法，但 mypy 默认禁止给方法赋值。只关闭这一条错误码，
 # attr-defined 等其余检查（含真实 bug）在测试代码中保持开启。
diff --git a/backend/src/main.py b/backend/src/main.py
index 57573ea7..1d56c447 100644
--- a/backend/src/main.py
+++ b/backend/src/main.py
@@ -7,12 +7,13 @@ import uvicorn
 from fastapi import APIRouter, Depends, FastAPI, Response
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse
+from starlette.routing import Route
 
 from module.api import v1
 from module.api.health import router as health_router
 from module.conf import VERSION, settings, setup_logger
 from module.core import AppContext
-from module.mcp import create_mcp_app
+from module.mcp import create_mcp_app, mcp
 
 setup_logger(reset=True)
 logger = logging.getLogger(__name__)
@@ -45,7 +46,9 @@ async def lifespan(app: FastAPI):
         # supervised background task and returns immediately (CONTRACT #5),
         # so this does not block server startup.
         await ctx.start_tasks()
-    yield
+    # 子应用不会运行自己的 lifespan，MCP 会话管理器在这里启动
+    async with app.state.mcp_sessions.run():
+        yield
     # Shutdown
     await ctx.stop()
     await ctx.plugins.stop()
@@ -71,8 +74,11 @@ def create_app() -> FastAPI:
     # it stays reachable without auth for container/orchestrator health checks
     app.include_router(health_router)
 
-    # mount MCP server (SSE transport for LLM tool integration)
-    app.mount("/mcp", create_mcp_app(ctx))
+    # MCP 服务（Streamable HTTP）：用路由而不是 mount，端点才是 /mcp 本身
+    # （mount 只匹配 /mcp/...，/mcp 会被 307 重定向到 /mcp/）
+    app.router.routes.append(Route("/mcp", endpoint=create_mcp_app(ctx)))
+    # 每次 create_mcp_app 都新建会话管理器，记下本应用的这一个供 lifespan 运行
+    app.state.mcp_sessions = mcp.session_manager
 
     return app
 
diff --git a/backend/src/module/mcp/__init__.py b/backend/src/module/mcp/__init__.py
index cdd0a305..7a994017 100644
--- a/backend/src/module/mcp/__init__.py
+++ b/backend/src/module/mcp/__init__.py
@@ -1,13 +1,16 @@
 """MCP (Model Context Protocol) server for AutoBangumi.
 
 Exposes anime subscriptions, RSS feeds, and download status to MCP clients
-(e.g. Claude Desktop) over a local-network-restricted SSE endpoint.
+(e.g. Claude Desktop) over a local-network-restricted Streamable HTTP endpoint.
 
 Usage::
 
     from module.mcp import create_mcp_app
 
-    app = create_mcp_app(ctx)  # returns a Starlette ASGI app, mount at /mcp
+    app = create_mcp_app(ctx)  # returns a Starlette ASGI app, route it at /mcp
 """
 
+# 导入即注册内置工具与资源
+from . import resources, tools  # noqa: F401
 from .server import create_mcp_starlette_app as create_mcp_app
+from .server import mcp
diff --git a/backend/src/module/mcp/plugins.py b/backend/src/module/mcp/plugins.py
index 1ca923b3..c34a7c8b 100644
--- a/backend/src/module/mcp/plugins.py
+++ b/backend/src/module/mcp/plugins.py
@@ -17,6 +17,7 @@ import re
 from collections.abc import Callable
 from typing import Any
 
+import jsonschema
 from mcp import types
 from pydantic import AnyUrl
 
@@ -78,7 +79,7 @@ def _tool_spec(entry: ProviderEntry, tool: McpTool) -> types.Tool:
     return types.Tool(
         name=tool_name(entry.plugin_id, entry.id),
         description=tool.description,
-        inputSchema=tool.input_schema,
+        input_schema=tool.input_schema,
     )
 
 
@@ -105,7 +106,7 @@ def list_plugin_tools() -> list[types.Tool]:
 async def call_plugin_tool(name: str, arguments: dict) -> tuple[bool, Any]:
     """调用插件工具；``name`` 不是插件工具时返回 ``(False, None)``。
 
-    插件处理函数的异常会继续抛出（由 ``handle_tool`` 统一转成错误 JSON），
+    插件处理函数的异常会继续抛出（由 ``AutoBangumiMCP.call_tool`` 统一转成错误结果），
     在此之前计入熔断。
     """
     if TOOL_SEPARATOR not in name:
@@ -114,6 +115,11 @@ async def call_plugin_tool(name: str, arguments: dict) -> tuple[bool, Any]:
     if found is None:
         return False, None
     entry, tool, _ = found
+    # mcp 2 的服务端不再按 inputSchema 校验参数；参数不合法是客户端的错，不计入熔断
+    try:
+        jsonschema.validate(arguments, tool.input_schema)
+    except jsonschema.ValidationError as e:
+        raise ValueError(f"Input validation error: {e.message}") from None
     try:
         result = await asyncio.wait_for(tool.handler(arguments), tool.timeout)
     except Exception as e:
@@ -137,10 +143,10 @@ def resource_uri(plugin_id: str, resource_id: str) -> str:
 
 def _resource_spec(entry: ProviderEntry, resource: McpResource) -> types.Resource:
     return types.Resource(
-        uri=AnyUrl(resource_uri(entry.plugin_id, entry.id)),
+        uri=str(AnyUrl(resource_uri(entry.plugin_id, entry.id))),
         name=resource.name,
         description=resource.description or None,
-        mimeType=resource.mime_type,
+        mime_type=resource.mime_type,
     )
 
 
@@ -148,7 +154,7 @@ def _plugin_resources() -> dict[str, tuple[ProviderEntry, McpResource, types.Res
     # 以 AnyUrl 规范化后的 URI 为键：含空格或非 ASCII 字符的 id 会被百分号编码，
     # 客户端读取时传回的正是列表里的编码形式
     return {
-        str(spec.uri): (entry, resource, spec)
+        spec.uri: (entry, resource, spec)
         for entry, resource, spec in _resolve(
             points.MCP_RESOURCE, McpResource, _resource_spec
         )
diff --git a/backend/src/module/mcp/resources.py b/backend/src/module/mcp/resources.py
index 120f4ceb..b441bb71 100644
--- a/backend/src/module/mcp/resources.py
+++ b/backend/src/module/mcp/resources.py
@@ -1,128 +1,82 @@
-"""MCP resource definitions and handlers for AutoBangumi.
+"""内置 MCP 资源：静态资源与 ``autobangumi://anime/{id}`` 模板，内容均为 JSON 文本。
 
-``RESOURCES`` lists static resources; ``RESOURCE_TEMPLATES`` lists URI
-templates for parameterised lookups. ``handle_resource`` resolves a URI
-to its JSON payload.
+插件提供的资源（``autobangumi://plugins/<plugin-id>/<id>``）由
+``AutoBangumiMCP.read_resource`` 分发。
 """
 
 import json
-import logging
 
-from mcp import types
-from pydantic import AnyUrl
-
-from module.conf import VERSION
 from module.database import Database
 from module.manager import TorrentManager
 from module.models import Bangumi
-from module.rss import RSSEngine
 
-from .plugins import list_plugin_resources, read_plugin_resource
-from .runtime import get_context
-from .tools import _bangumi_to_dict
+from .server import mcp
+from .tools import _bangumi_to_dict, get_program_status
 
-logger = logging.getLogger(__name__)
-
-RESOURCES = [
-    types.Resource(
-        uri=AnyUrl("autobangumi://anime/list"),
-        name="All tracked anime",
-        description="List of all anime subscriptions being tracked by AutoBangumi",
-        mimeType="application/json",
-    ),
-    types.Resource(
-        uri=AnyUrl("autobangumi://status"),
-        name="Program status",
-        description="Current AutoBangumi program status, version, and state",
-        mimeType="application/json",
-    ),
-    types.Resource(
-        uri=AnyUrl("autobangumi://rss/feeds"),
-        name="RSS feeds",
-        description="All configured RSS feeds with health status",
-        mimeType="application/json",
-    ),
-]
-
-RESOURCE_TEMPLATES = [
-    types.ResourceTemplate(
-        uriTemplate="autobangumi://anime/{id}",
-        name="Anime details",
-        description="Detailed information about a specific tracked anime by ID",
-        mimeType="application/json",
-    ),
-]
+JSON = "application/json"
 
 
-def all_resources() -> list[types.Resource]:
-    """内置资源 + 当前已启用插件提供的资源（``autobangumi://plugins/<id>/...``）。"""
-    return RESOURCES + list_plugin_resources()
+@mcp.resource(
+    "autobangumi://anime/list",
+    name="All tracked anime",
+    description="List of all anime subscriptions being tracked by AutoBangumi",
+    mime_type=JSON,
+)
+async def anime_list() -> str:
+    async with Database() as db:
+        items = await db.bangumi.search_all()
+    return json.dumps([_bangumi_to_dict(b) for b in items], ensure_ascii=False)
 
 
-async def handle_resource(uri: str) -> str:
-    """Return a JSON string for the given MCP resource URI.
+@mcp.resource(
+    "autobangumi://status",
+    name="Program status",
+    description="Current AutoBangumi program status, version, and state",
+    mime_type=JSON,
+)
+def program_status() -> str:
+    return json.dumps(get_program_status())
 
-    Supported URIs:
-    - ``autobangumi://anime/list`` - all tracked anime
-    - ``autobangumi://status`` - program version and running state
-    - ``autobangumi://rss/feeds`` - configured RSS feeds
-    - ``autobangumi://anime/{id}`` - single anime by integer ID
-    - ``autobangumi://plugins/<plugin-id>/<id>`` - 插件提供的资源
-    """
-    if uri == "autobangumi://anime/list":
-        async with Database() as db:
-            items = await db.bangumi.search_all()
-        return json.dumps([_bangumi_to_dict(b) for b in items], ensure_ascii=False)
 
-    elif uri == "autobangumi://status":
-        ctx = get_context()
-        return json.dumps(
+@mcp.resource(
+    "autobangumi://rss/feeds",
+    name="RSS feeds",
+    description="All configured RSS feeds with health status",
+    mime_type=JSON,
+)
+async def rss_feeds() -> str:
+    async with Database() as db:
+        feeds = await db.rss.search_all()
+    return json.dumps(
+        [
             {
-                "version": VERSION,
-                "running": ctx.is_running if ctx is not None else False,
-                "first_run": ctx.first_run if ctx is not None else True,
+                "id": f.id,
+                "name": f.name,
+                "url": f.url,
+                "enabled": f.enabled,
+                "connection_status": f.connection_status,
+                "last_checked_at": f.last_checked_at,
             }
-        )
+            for f in feeds
+        ],
+        ensure_ascii=False,
+    )
 
-    elif uri == "autobangumi://rss/feeds":
-        async with Database() as db:
-            feeds = await db.rss.search_all()
-        return json.dumps(
-            [
-                {
-                    "id": f.id,
-                    "name": f.name,
-                    "url": f.url,
-                    "enabled": f.enabled,
-                    "connection_status": f.connection_status,
-                    "last_checked_at": f.last_checked_at,
-                }
-                for f in feeds
-            ],
-            ensure_ascii=False,
-        )
-
-    elif uri.startswith("autobangumi://anime/"):
-        anime_id_str = uri.split("/")[-1]
-        try:
-            anime_id = int(anime_id_str)
-        except ValueError:
-            return json.dumps({"error": f"Invalid anime ID: {anime_id_str}"})
-        async with Database() as db:
-            manager = TorrentManager(db)
-            result = await manager.search_one(anime_id)
-        if isinstance(result, Bangumi):
-            return json.dumps(_bangumi_to_dict(result), ensure_ascii=False)
-        return json.dumps({"error": result.msg_en})
 
+@mcp.resource(
+    "autobangumi://anime/{id}",
+    name="Anime details",
+    description="Detailed information about a specific tracked anime by ID",
+    mime_type=JSON,
+)
+async def anime_detail(id: str) -> str:
     try:
-        handled, content = await read_plugin_resource(uri)
-    except Exception as e:
-        logger.warning("Plugin resource %s failed: %s", uri, e)
-        return json.dumps({"error": str(e)}, ensure_ascii=False)
-    if handled:
-        if isinstance(content, str):
-            return content
-        return json.dumps(content, ensure_ascii=False, default=str)
-
-    return json.dumps({"error": f"Unknown resource: {uri}"})
+        anime_id = int(id)
+    except ValueError:
+        return json.dumps({"error": f"Invalid anime ID: {id}"})
+    async with Database() as db:
+        manager = TorrentManager(db)
+        result = await manager.search_one(anime_id)
+    if isinstance(result, Bangumi):
+        return json.dumps(_bangumi_to_dict(result), ensure_ascii=False)
+    return json.dumps({"error": result.msg_en})
diff --git a/backend/src/module/mcp/server.py b/backend/src/module/mcp/server.py
index b05de20a..b998aaa1 100644
--- a/backend/src/module/mcp/server.py
+++ b/backend/src/module/mcp/server.py
@@ -1,99 +1,115 @@
 """MCP server assembly for AutoBangumi.
 
-Wires together the MCP ``Server``, SSE transport, tool/resource handlers,
-and local-network middleware into a single Starlette ASGI application.
+``mcp`` 是 :class:`AutoBangumiMCP`（``MCPServer`` 子类）的单例：内置工具与资源
+在 ``tools.py`` / ``resources.py`` 中用 ``@mcp.tool()`` / ``@mcp.resource()``
+注册，插件提供的工具与资源在本类的 list / call / read 中追加与分发。
 
-Mount the app returned by ``create_mcp_starlette_app`` at a path prefix
-(e.g. ``/mcp``) in the parent FastAPI application to expose the MCP
-endpoint at ``/mcp/sse``.
+``create_mcp_starlette_app`` 返回 Streamable HTTP 传输的 Starlette 应用，
+由 ``main.py`` 以 ``/mcp`` 路由挂载。
 """
 
+import json
 import logging
+from collections.abc import Iterable
+from typing import Any
 
 from mcp import types
-from mcp.server import Server
-from mcp.server.sse import SseServerTransport
+from mcp.server.lowlevel.helper_types import ReadResourceContents
+from mcp.server.mcpserver import Context, MCPServer
+from mcp.server.mcpserver.exceptions import ResourceError
+from mcp.server.transport_security import TransportSecuritySettings
 from pydantic import AnyUrl
 from starlette.applications import Starlette
-from starlette.routing import Mount, Route
-from starlette.types import Receive, Scope, Send
 
-from .resources import RESOURCE_TEMPLATES, all_resources, handle_resource
+from .plugins import (
+    call_plugin_tool,
+    list_plugin_resources,
+    list_plugin_tools,
+    read_plugin_resource,
+)
 from .runtime import set_context
 from .security import McpAccessMiddleware
-from .tools import all_tools, handle_tool
 
 logger = logging.getLogger(__name__)
 
-server = Server("autobangumi")
-sse = SseServerTransport("/messages/")
+
+def _text(value: Any) -> types.TextContent:
+    return types.TextContent(
+        type="text", text=json.dumps(value, ensure_ascii=False, default=str)
+    )
 
 
-@server.list_tools()
-async def list_tools() -> list[types.Tool]:
-    return all_tools()
+class AutoBangumiMCP(MCPServer):
+    """内置工具与资源走 MCPServer 的注册表，插件的从扩展注册表现取。"""
 
+    async def list_tools(self) -> list[types.Tool]:
+        return await super().list_tools() + list_plugin_tools()
 
-@server.call_tool()
-async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
-    logger.debug("Tool called: %s", name)
-    return await handle_tool(name, arguments)
-
-
-@server.list_resources()
-async def list_resources() -> list[types.Resource]:
-    return all_resources()
-
-
-@server.list_resource_templates()
-async def list_resource_templates() -> list[types.ResourceTemplate]:
-    return RESOURCE_TEMPLATES
-
-
-@server.read_resource()
-async def read_resource(uri: AnyUrl) -> str:
-    logger.debug("Resource read: %s", uri)
-    # 底层 read_resource 传入的是 pydantic AnyUrl，handle_resource 按字符串匹配
-    return await handle_resource(str(uri))
-
-
-class _SseEndpoint:
-    """``GET /sse`` 的裸 ASGI 端点：建立 SSE 连接并运行 MCP 会话直到客户端断开。
-
-    ``connect_sse`` 自己经 ``send`` 写出整个 HTTP 响应，所以这里不能是返回
-    ``Response`` 的普通 endpoint 函数——Starlette 会在 SSE 结束后再发一次
-    ``http.response.start``，每次断开都触发 ``AssertionError``。
-    """
-
-    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
-        async with sse.connect_sse(scope, receive, send) as streams:
-            await server.run(
-                streams[0],
-                streams[1],
-                server.create_initialization_options(),
+    async def call_tool(
+        self,
+        name: str,
+        arguments: dict[str, Any],
+        context: Context | None = None,
+    ) -> types.CallToolResult:
+        # 结果统一序列化为一段 JSON 文本；内置工具注册时关闭 structured_output，
+        # 不声明 outputSchema
+        logger.debug("Tool called: %s", name)
+        try:
+            handled, result = await call_plugin_tool(name, arguments)
+            if not handled:
+                result = await self._tool_manager.call_tool(
+                    name, arguments, context or Context(mcp_server=self)
+                )
+        except Exception as e:
+            logger.exception("Tool %s failed", name)
+            # MCPServer 把工具抛出的异常包成 ToolError，原始异常在 __cause__
+            error = e.__cause__ or e
+            return types.CallToolResult(
+                content=[_text({"error": str(error)})], is_error=True
             )
+        return types.CallToolResult(content=[_text(result)])
+
+    async def list_resources(self) -> list[types.Resource]:
+        return await super().list_resources() + list_plugin_resources()
+
+    async def read_resource(
+        self, uri: AnyUrl | str, context: Context | None = None
+    ) -> Iterable[ReadResourceContents] | types.InputRequiredResult:
+        logger.debug("Resource read: %s", uri)
+        try:
+            handled, content = await read_plugin_resource(str(uri))
+        except Exception as e:
+            logger.warning("Plugin resource %s failed: %s", uri, e)
+            raise ResourceError(str(e) or type(e).__name__) from e
+        if not handled:
+            return await super().read_resource(uri, context)
+        if not isinstance(content, str):
+            content = json.dumps(content, ensure_ascii=False, default=str)
+        return [ReadResourceContents(content=content, mime_type=None)]
 
 
-handle_sse = _SseEndpoint()
+mcp = AutoBangumiMCP("autobangumi")
 
 
 def create_mcp_starlette_app(ctx=None) -> Starlette:
     """Build and return the MCP Starlette sub-application.
 
-    Routes:
-    - ``GET /sse`` - SSE stream for MCP clients
-    - ``POST /messages/`` - client-to-server message posting
+    只有一个 Streamable HTTP 端点 ``/mcp``（GET / POST / DELETE）。每次调用都
+    新建会话管理器，调用方须在应用 lifespan 中运行 ``mcp.session_manager.run()``。
 
     ``ctx`` is the application :class:`AppContext`; it is stored so status
     tools/resources can report live program state. ``McpAccessMiddleware`` is
     applied to enforce configurable IP whitelist and bearer token access control.
     """
     set_context(ctx)
-    app = Starlette(
-        routes=[
-            Route("/sse", endpoint=handle_sse),
-            Mount("/messages", app=sse.handle_post_message),
-        ],
+    app = mcp.streamable_http_app(
+        streamable_http_path="/mcp",
+        # 默认 host=127.0.0.1 会开启 DNS rebinding 防护并只放行 localhost 的
+        # Host 头，局域网地址与反向代理域名都会被拒（421）。访问控制由
+        # McpAccessMiddleware 的 IP 白名单与令牌负责，与原 SSE 端点一致
+        transport_security=TransportSecuritySettings(
+            enable_dns_rebinding_protection=False
+        ),
     )
     app.add_middleware(McpAccessMiddleware)
     return app
diff --git a/backend/src/module/mcp/tools.py b/backend/src/module/mcp/tools.py
index 551f8e20..84238832 100644
--- a/backend/src/module/mcp/tools.py
+++ b/backend/src/module/mcp/tools.py
@@ -1,8 +1,13 @@
-import json
-import logging
-from typing import Any
+"""内置 MCP 工具：参数 schema 由类型注解生成。
 
-from mcp import types
+``structured_output=False``：结果由 ``AutoBangumiMCP.call_tool`` 序列化为一段
+JSON 文本，不声明 outputSchema。
+"""
+
+import json
+from typing import Annotated, Literal
+
+from pydantic import Field
 
 from module.conf import VERSION
 from module.database import Database
@@ -12,174 +17,10 @@ from module.models import Bangumi, BangumiUpdate, RSSItem
 from module.rss import RSSAnalyser, RSSEngine
 from module.searcher import SearchTorrent
 
-from .plugins import call_plugin_tool, list_plugin_tools
 from .runtime import get_context
+from .server import mcp
 
-logger = logging.getLogger(__name__)
-
-TOOLS = [
-    types.Tool(
-        name="list_anime",
-        description="List all tracked anime subscriptions. Returns title, season, status, and episode offset for each.",
-        inputSchema={
-            "type": "object",
-            "properties": {
-                "active_only": {
-                    "type": "boolean",
-                    "description": "If true, only return active (non-disabled) anime",
-                    "default": False,
-                },
-            },
-        },
-    ),
-    types.Tool(
-        name="get_anime",
-        description="Get detailed information about a specific anime subscription by its ID.",
-        inputSchema={
-            "type": "object",
-            "properties": {
-                "id": {
-                    "type": "integer",
-                    "description": "The anime/bangumi ID",
-                },
-            },
-            "required": ["id"],
-        },
-    ),
-    types.Tool(
-        name="search_anime",
-        description="Search for anime torrents across torrent sites (Mikan, DMHY, Nyaa). Returns available anime matching the keywords.",
-        inputSchema={
-            "type": "object",
-            "properties": {
-                "keywords": {
-                    "type": "string",
-                    "description": "Search keywords (e.g. anime title)",
-                },
-                "site": {
-                    "type": "string",
-                    "description": "Torrent site to search",
-                    "enum": ["mikan", "dmhy", "nyaa"],
-                    "default": "mikan",
-                },
-            },
-            "required": ["keywords"],
-        },
-    ),
-    types.Tool(
-        name="subscribe_anime",
-        description="Subscribe to an anime series by providing its RSS link. Analyzes the RSS feed and sets up automatic downloading.",
-        inputSchema={
-            "type": "object",
-            "properties": {
-                "rss_link": {
-                    "type": "string",
-                    "description": "RSS feed URL for the anime (obtained from search_anime results)",
-                },
-                "parser": {
-                    "type": "string",
-                    "description": "RSS parser type",
-                    "enum": ["mikan", "dmhy", "nyaa"],
-                    "default": "mikan",
-                },
-            },
-            "required": ["rss_link"],
-        },
-    ),
-    types.Tool(
-        name="unsubscribe_anime",
-        description="Unsubscribe from an anime. Can either disable (keeps data) or fully delete the subscription.",
-        inputSchema={
-            "type": "object",
-            "properties": {
-                "id": {
-                    "type": "integer",
-                    "description": "The anime/bangumi ID to unsubscribe",
-                },
-                "delete": {
-                    "type": "boolean",
-                    "description": "If true, permanently delete the subscription. If false, just disable it.",
-                    "default": False,
-                },
-            },
-            "required": ["id"],
-        },
-    ),
-    types.Tool(
-        name="list_downloads",
-        description="Show current torrent download status from the download client (qBittorrent/Aria2).",
-        inputSchema={
-            "type": "object",
-            "properties": {
-                "status": {
-                    "type": "string",
-                    "description": "Filter by download status",
-                    "enum": ["all", "downloading", "completed", "paused"],
-                    "default": "all",
-                },
-            },
-        },
-    ),
-    types.Tool(
-        name="list_rss_feeds",
-        description="List all configured RSS feeds with their connection status and health information.",
-        inputSchema={
-            "type": "object",
-            "properties": {},
-        },
-    ),
-    types.Tool(
-        name="get_program_status",
-        description="Get the current program status including version, running state, and first-run flag.",
-        inputSchema={
-            "type": "object",
-            "properties": {},
-        },
-    ),
-    types.Tool(
-        name="refresh_feeds",
-        description="Trigger an immediate refresh of all RSS feeds to check for new episodes.",
-        inputSchema={
-            "type": "object",
-            "properties": {},
-        },
-    ),
-    types.Tool(
-        name="update_anime",
-        description="Update settings for a tracked anime (episode offset, season offset, filters, etc.).",
-        inputSchema={
-            "type": "object",
-            "properties": {
-                "id": {
-                    "type": "integer",
-                    "description": "The anime/bangumi ID to update",
-                },
-                "episode_offset": {
-                    "type": "integer",
-                    "description": "Episode number offset for renaming",
-                },
-                "season_offset": {
-                    "type": "integer",
-                    "description": "Season number offset for renaming",
-                },
-                "season": {
-                    "type": "integer",
-                    "description": "Season number",
-                },
-                "filter": {
-                    "type": "string",
-                    "description": "Comma-separated filter patterns to exclude",
-                },
-            },
-            "required": ["id"],
-        },
-    ),
-]
-
-
-def all_tools() -> list[types.Tool]:
-    """内置工具 + 当前已启用插件提供的工具（``<plugin-id>__<id>``）。"""
-    return TOOLS + list_plugin_tools()
+Site = Literal["mikan", "dmhy", "nyaa"]
 
 
 def _bangumi_to_dict(b: Bangumi) -> dict:
@@ -205,51 +46,16 @@ def _bangumi_to_dict(b: Bangumi) -> dict:
     }
 
 
-async def handle_tool(name: str, arguments: dict) -> list[types.TextContent]:
-    try:
-        result = await _dispatch(name, arguments)
-        return [
-            types.TextContent(
-                type="text", text=json.dumps(result, ensure_ascii=False, default=str)
-            )
-        ]
-    except Exception as e:
-        logger.exception("Tool %s failed", name)
-        return [
-            types.TextContent(
-                type="text", text=json.dumps({"error": str(e)}, ensure_ascii=False)
-            )
-        ]
-
-
-async def _dispatch(name: str, args: dict) -> Any:
-    if name == "list_anime":
-        return await _list_anime(args.get("active_only", False))
-    elif name == "get_anime":
-        return await _get_anime(args["id"])
-    elif name == "search_anime":
-        return await _search_anime(args["keywords"], args.get("site", "mikan"))
-    elif name == "subscribe_anime":
-        return await _subscribe_anime(args["rss_link"], args.get("parser", "mikan"))
-    elif name == "unsubscribe_anime":
-        return await _unsubscribe_anime(args["id"], args.get("delete", False))
-    elif name == "list_downloads":
-        return await _list_downloads(args.get("status", "all"))
-    elif name == "list_rss_feeds":
-        return await _list_rss_feeds()
-    elif name == "get_program_status":
-        return _get_program_status()
-    elif name == "refresh_feeds":
-        return await _refresh_feeds()
-    elif name == "update_anime":
-        return await _update_anime(args)
-    handled, result = await call_plugin_tool(name, args)
-    if handled:
-        return result
-    return {"error": f"Unknown tool: {name}"}
-
-
-async def _list_anime(active_only: bool) -> list[dict]:
+@mcp.tool(
+    description="List all tracked anime subscriptions. Returns title, season, status, and episode offset for each.",
+    structured_output=False,
+)
+async def list_anime(
+    active_only: Annotated[
+        bool,
+        Field(description="If true, only return active (non-disabled) anime"),
+    ] = False,
+) -> list[dict]:
     async with Database() as db:
         manager = TorrentManager(db)
         if active_only:
@@ -259,16 +65,29 @@ async def _list_anime(active_only: bool) -> list[dict]:
     return [_bangumi_to_dict(b) for b in items]
 
 
-async def _get_anime(bangumi_id: int) -> dict:
+@mcp.tool(
+    description="Get detailed information about a specific anime subscription by its ID.",
+    structured_output=False,
+)
+async def get_anime(
+    id: Annotated[int, Field(description="The anime/bangumi ID")],
+) -> dict:
     async with Database() as db:
         manager = TorrentManager(db)
-        result = await manager.search_one(bangumi_id)
+        result = await manager.search_one(id)
     if isinstance(result, Bangumi):
         return _bangumi_to_dict(result)
     return {"error": result.msg_en}
 
 
-async def _search_anime(keywords: str, site: str) -> list[dict]:
+@mcp.tool(
+    description="Search for anime torrents across torrent sites (Mikan, DMHY, Nyaa). Returns available anime matching the keywords.",
+    structured_output=False,
+)
+async def search_anime(
+    keywords: Annotated[str, Field(description="Search keywords (e.g. anime title)")],
+    site: Annotated[Site, Field(description="Torrent site to search")] = "mikan",
+) -> list[dict]:
     keyword_list = keywords.split()
     results = []
     st = SearchTorrent()
@@ -279,7 +98,19 @@ async def _search_anime(keywords: str, site: str) -> list[dict]:
     return results
 
 
-async def _subscribe_anime(rss_link: str, parser: str) -> dict:
+@mcp.tool(
+    description="Subscribe to an anime series by providing its RSS link. Analyzes the RSS feed and sets up automatic downloading.",
+    structured_output=False,
+)
+async def subscribe_anime(
+    rss_link: Annotated[
+        str,
+        Field(
+            description="RSS feed URL for the anime (obtained from search_anime results)"
+        ),
+    ],
+    parser: Annotated[Site, Field(description="RSS parser type")] = "mikan",
+) -> dict:
     analyser = RSSAnalyser()
     rss = RSSItem(url=rss_link, parser=parser)
     data = await analyser.link_to_data(rss)
@@ -289,17 +120,38 @@ async def _subscribe_anime(rss_link: str, parser: str) -> dict:
     return {"status": resp.status, "message": resp.msg_en}
 
 
-async def _unsubscribe_anime(bangumi_id: int, delete: bool) -> dict:
+@mcp.tool(
+    description="Unsubscribe from an anime. Can either disable (keeps data) or fully delete the subscription.",
+    structured_output=False,
+)
+async def unsubscribe_anime(
+    id: Annotated[int, Field(description="The anime/bangumi ID to unsubscribe")],
+    delete: Annotated[
+        bool,
+        Field(
+            description="If true, permanently delete the subscription. If false, just disable it."
+        ),
+    ] = False,
+) -> dict:
     async with Database() as db:
         manager = TorrentManager(db)
         if delete:
-            resp = await manager.delete_rule(bangumi_id)
+            resp = await manager.delete_rule(id)
         else:
-            resp = await manager.disable_rule(bangumi_id)
+            resp = await manager.disable_rule(id)
     return {"status": resp.status, "message": resp.msg_en}
 
 
-async def _list_downloads(status: str) -> list[dict]:
+@mcp.tool(
+    description="Show current torrent download status from the download client (qBittorrent/Aria2).",
+    structured_output=False,
+)
+async def list_downloads(
+    status: Annotated[
+        Literal["all", "downloading", "completed", "paused"],
+        Field(description="Filter by download status"),
+    ] = "all",
+) -> list[dict]:
     status_filter = None if status == "all" else status
     torrents = await list_torrents(status_filter) or []
     return [
@@ -317,7 +169,11 @@ async def _list_downloads(status: str) -> list[dict]:
     ]
 
 
-async def _list_rss_feeds() -> list[dict]:
+@mcp.tool(
+    description="List all configured RSS feeds with their connection status and health information.",
+    structured_output=False,
+)
+async def list_rss_feeds() -> list[dict]:
     async with Database() as db:
         feeds = await db.rss.search_all()
     return [
@@ -336,7 +192,11 @@ async def _list_rss_feeds() -> list[dict]:
     ]
 
 
-def _get_program_status() -> dict:
+@mcp.tool(
+    description="Get the current program status including version, running state, and first-run flag.",
+    structured_output=False,
+)
+def get_program_status() -> dict:
     ctx = get_context()
     return {
         "version": VERSION,
@@ -345,7 +205,11 @@ def _get_program_status() -> dict:
     }
 
 
-async def _refresh_feeds() -> dict:
+@mcp.tool(
+    description="Trigger an immediate refresh of all RSS feeds to check for new episodes.",
+    structured_output=False,
+)
+async def refresh_feeds() -> dict:
     async with DownloaderPool() as downloaders:
         async with Database() as db:
             engine = RSSEngine(db)
@@ -353,23 +217,38 @@ async def _refresh_feeds() -> dict:
     return {"status": True, "message": "RSS feeds refreshed successfully"}
 
 
-async def _update_anime(args: dict) -> dict:
-    bangumi_id = args["id"]
+@mcp.tool(
+    description="Update settings for a tracked anime (episode offset, season offset, filters, etc.).",
+    structured_output=False,
+)
+async def update_anime(
+    id: Annotated[int, Field(description="The anime/bangumi ID to update")],
+    episode_offset: Annotated[
+        int | None, Field(description="Episode number offset for renaming")
+    ] = None,
+    season_offset: Annotated[
+        int | None, Field(description="Season number offset for renaming")
+    ] = None,
+    season: Annotated[int | None, Field(description="Season number")] = None,
+    filter: Annotated[
+        str | None, Field(description="Comma-separated filter patterns to exclude")
+    ] = None,
+) -> dict:
     async with Database() as db:
         manager = TorrentManager(db)
-        existing = await db.bangumi.search_id(bangumi_id)
+        existing = await db.bangumi.search_id(id)
         if not existing:
-            return {"error": f"Anime with id {bangumi_id} not found"}
+            return {"error": f"Anime with id {id} not found"}
 
         update_data = BangumiUpdate(**existing.model_dump())
-        if "episode_offset" in args:
-            update_data.episode_offset = args["episode_offset"]
-        if "season_offset" in args:
-            update_data.season_offset = args["season_offset"]
-        if "season" in args:
-            update_data.season = args["season"]
-        if "filter" in args:
-            update_data.filter = args["filter"]
+        if episode_offset is not None:
+            update_data.episode_offset = episode_offset
+        if season_offset is not None:
+            update_data.season_offset = season_offset
+        if season is not None:
+            update_data.season = season
+        if filter is not None:
+            update_data.filter = filter
 
-        resp = await manager.update_rule(bangumi_id, update_data)
+        resp = await manager.update_rule(id, update_data)
     return {"status": resp.status, "message": resp.msg_en}
diff --git a/backend/src/test/test_main_lifespan.py b/backend/src/test/test_main_lifespan.py
index 42fd3fc7..41b17ae5 100644
--- a/backend/src/test/test_main_lifespan.py
+++ b/backend/src/test/test_main_lifespan.py
@@ -64,3 +64,59 @@ class TestLifespan:
                     pass
         # 迁移失败时插件不应启动（插件 KV 表由迁移创建）
         ctx.plugins.start.assert_not_called()
+
+
+class TestMcpEndpoint:
+    """MCP 走 Streamable HTTP，端点就是 /mcp；会话管理器由应用 lifespan 启动。"""
+
+    HEADERS = {
+        "Authorization": "Bearer mcp-token",
+        "Accept": "application/json, text/event-stream",
+    }
+
+    def test_streamable_http_initialize_and_list_tools(self):
+        service = MagicMock()
+        service.authenticate_api_token = AsyncMock(return_value=MagicMock())
+        with (
+            patch("main.AppContext.build", return_value=_lifespan_ctx()),
+            patch("module.mcp.security.auth_service", service),
+        ):
+            app = main.create_app()
+            with TestClient(app, follow_redirects=False) as client:
+                init = client.post(
+                    "/mcp",
+                    headers=self.HEADERS,
+                    json={
+                        "jsonrpc": "2.0",
+                        "id": 1,
+                        "method": "initialize",
+                        "params": {
+                            "protocolVersion": "2025-06-18",
+                            "capabilities": {},
+                            "clientInfo": {"name": "test", "version": "0"},
+                        },
+                    },
+                )
+                assert init.status_code == 200
+                session = {"mcp-session-id": init.headers["mcp-session-id"]}
+                client.post(
+                    "/mcp",
+                    headers=self.HEADERS | session,
+                    json={"jsonrpc": "2.0", "method": "notifications/initialized"},
+                )
+                listed = client.post(
+                    "/mcp",
+                    headers=self.HEADERS | session,
+                    json={"jsonrpc": "2.0", "id": 2, "method": "tools/list"},
+                )
+                assert listed.status_code == 200
+                assert '"list_anime"' in listed.text
+                # 旧的 SSE 端点已删除
+                assert client.get("/mcp/sse", headers=self.HEADERS).status_code == 404
+
+    def test_rejected_without_token_or_whitelisted_ip(self):
+        with patch("main.AppContext.build", return_value=_lifespan_ctx()):
+            app = main.create_app()
+            with TestClient(app) as client:
+                resp = client.post("/mcp", json={})
+        assert resp.status_code == 403
diff --git a/backend/src/test/test_mcp_resources.py b/backend/src/test/test_mcp_resources.py
index 9473f46c..514fd325 100644
--- a/backend/src/test/test_mcp_resources.py
+++ b/backend/src/test/test_mcp_resources.py
@@ -1,16 +1,14 @@
-"""Tests for module.mcp.resources - await handle_resource() and _bangumi_to_dict()."""
+"""Tests for module.mcp.resources - reads via mcp.read_resource() and _bangumi_to_dict()."""
 
 import json
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
+from mcp.server.lowlevel.helper_types import ReadResourceContents
+from mcp.server.mcpserver.exceptions import ResourceNotFoundError
 
-from module.mcp.resources import (
-    RESOURCE_TEMPLATES,
-    RESOURCES,
-    _bangumi_to_dict,
-    handle_resource,
-)
+from module.mcp import mcp
+from module.mcp.tools import _bangumi_to_dict
 from module.models import Bangumi, ResponseModel
 from test.factories import make_bangumi
 
@@ -53,36 +51,38 @@ def _mock_sync_manager(bangumi_list=None, single=None):
     return mock_mgr, mock_mgr
 
 
+async def handle_resource(uri: str) -> str:
+    """经 ``mcp.read_resource`` 读取资源，返回文本内容。"""
+    [content] = await mcp.read_resource(uri)
+    assert isinstance(content, ReadResourceContents)
+    assert isinstance(content.content, str)
+    return content.content
+
+
 def _parse(raw: str) -> dict | list:
     return json.loads(raw)
 
 
 # ---------------------------------------------------------------------------
-# Static metadata (RESOURCES / RESOURCE_TEMPLATES)
+# Static metadata (list_resources / list_resource_templates)
 # ---------------------------------------------------------------------------
 
 
 class TestResourceMetadata:
-    """Verify the static resource and template lists."""
-
-    async def test_resources_is_list(self):
-        assert isinstance(RESOURCES, list)
-
-    async def test_resources_not_empty(self):
-        assert len(RESOURCES) > 0
+    """Verify the registered resources and templates."""
 
     async def test_resource_uris_present(self):
-        uris = {str(r.uri) for r in RESOURCES}
-        assert "autobangumi://anime/list" in uris
-        assert "autobangumi://status" in uris
-        assert "autobangumi://rss/feeds" in uris
-
-    async def test_resource_templates_is_list(self):
-        assert isinstance(RESOURCE_TEMPLATES, list)
+        resources = await mcp.list_resources()
+        assert {r.uri for r in resources} == {
+            "autobangumi://anime/list",
+            "autobangumi://status",
+            "autobangumi://rss/feeds",
+        }
+        assert all(r.mime_type == "application/json" for r in resources)
 
     async def test_anime_id_template_present(self):
-        templates = [str(t.uriTemplate) for t in RESOURCE_TEMPLATES]
-        assert "autobangumi://anime/{id}" in templates
+        templates = [t.uri_template for t in await mcp.list_resource_templates()]
+        assert templates == ["autobangumi://anime/{id}"]
 
 
 # ---------------------------------------------------------------------------
@@ -214,8 +214,8 @@ class TestHandleResourceStatus:
 
     async def test_returns_json_string(self, mock_program):
         with (
-            patch("module.mcp.resources.VERSION", "3.2.0-test"),
-            patch("module.mcp.resources.get_context", return_value=mock_program),
+            patch("module.mcp.tools.VERSION", "3.2.0-test"),
+            patch("module.mcp.tools.get_context", return_value=mock_program),
         ):
             raw = await handle_resource("autobangumi://status")
         assert isinstance(raw, str)
@@ -223,8 +223,8 @@ class TestHandleResourceStatus:
 
     async def test_version_in_output(self, mock_program):
         with (
-            patch("module.mcp.resources.VERSION", "3.2.0-test"),
-            patch("module.mcp.resources.get_context", return_value=mock_program),
+            patch("module.mcp.tools.VERSION", "3.2.0-test"),
+            patch("module.mcp.tools.get_context", return_value=mock_program),
         ):
             result = _parse(await handle_resource("autobangumi://status"))
         assert isinstance(result, dict)
@@ -233,8 +233,8 @@ class TestHandleResourceStatus:
     async def test_running_true(self, mock_program):
         mock_program.is_running = True
         with (
-            patch("module.mcp.resources.VERSION", "3.2.0-test"),
-            patch("module.mcp.resources.get_context", return_value=mock_program),
+            patch("module.mcp.tools.VERSION", "3.2.0-test"),
+            patch("module.mcp.tools.get_context", return_value=mock_program),
         ):
             result = _parse(await handle_resource("autobangumi://status"))
         assert isinstance(result, dict)
@@ -243,8 +243,8 @@ class TestHandleResourceStatus:
     async def test_first_run_false(self, mock_program):
         mock_program.first_run = False
         with (
-            patch("module.mcp.resources.VERSION", "3.2.0-test"),
-            patch("module.mcp.resources.get_context", return_value=mock_program),
+            patch("module.mcp.tools.VERSION", "3.2.0-test"),
+            patch("module.mcp.tools.get_context", return_value=mock_program),
         ):
             result = _parse(await handle_resource("autobangumi://status"))
         assert isinstance(result, dict)
@@ -252,8 +252,8 @@ class TestHandleResourceStatus:
 
     async def test_all_keys_present(self, mock_program):
         with (
-            patch("module.mcp.resources.VERSION", "3.2.0-test"),
-            patch("module.mcp.resources.get_context", return_value=mock_program),
+            patch("module.mcp.tools.VERSION", "3.2.0-test"),
+            patch("module.mcp.tools.get_context", return_value=mock_program),
         ):
             result = _parse(await handle_resource("autobangumi://status"))
         assert isinstance(result, dict)
@@ -374,26 +374,15 @@ class TestHandleResourceAnimeById:
 class TestHandleResourceUnknown:
     """Tests for unrecognised resource URIs."""
 
-    async def test_unknown_uri_returns_json_error(self):
-        """An unrecognised URI produces a JSON object with 'error'."""
-        result = _parse(await handle_resource("autobangumi://does/not/exist"))
-        assert "error" in result
-
-    async def test_unknown_uri_mentions_uri_in_error(self):
-        """The error message includes the unrecognised URI."""
-        uri = "autobangumi://does/not/exist"
-        result = _parse(await handle_resource(uri))
-        assert isinstance(result, dict)
-        assert uri in result["error"]
-
-    async def test_empty_uri_returns_error(self):
-        """An empty string URI returns a JSON error."""
-        result = _parse(await handle_resource(""))
-        assert "error" in result
-
-    async def test_completely_different_scheme_returns_error(self):
-        """A URI with a wrong scheme returns a JSON error."""
-        result = _parse(
-            await handle_resource("https://autobangumi.example.com/anime/list")
-        )
-        assert "error" in result
+    @pytest.mark.parametrize(
+        "uri",
+        [
+            "autobangumi://does/not/exist",
+            "",
+            "https://autobangumi.example.com/anime/list",
+        ],
+    )
+    async def test_unknown_uri_raises_not_found(self, uri):
+        """未知 URI 抛出 ResourceNotFoundError（客户端收到 JSON-RPC 错误），消息带上 URI。"""
+        with pytest.raises(ResourceNotFoundError, match=f"Unknown resource: {uri}"):
+            await mcp.read_resource(uri)
diff --git a/backend/src/test/test_mcp_runtime.py b/backend/src/test/test_mcp_runtime.py
index badeca50..80f60c3e 100644
--- a/backend/src/test/test_mcp_runtime.py
+++ b/backend/src/test/test_mcp_runtime.py
@@ -4,7 +4,7 @@ from unittest.mock import MagicMock
 
 from module.mcp import create_mcp_app
 from module.mcp.runtime import get_context, set_context
-from module.mcp.tools import _get_program_status
+from module.mcp.tools import get_program_status
 
 
 class TestRuntimeContext:
@@ -28,7 +28,7 @@ class TestProgramStatusReadsContext:
         ctx.first_run = False
         set_context(ctx)
         try:
-            result = _get_program_status()
+            result = get_program_status()
         finally:
             set_context(None)
         assert result["running"] is True
@@ -36,6 +36,6 @@ class TestProgramStatusReadsContext:
 
     def test_status_without_context_is_safe(self):
         set_context(None)
-        result = _get_program_status()
+        result = get_program_status()
         assert result["running"] is False
         assert result["first_run"] is True
diff --git a/backend/src/test/test_mcp_tools.py b/backend/src/test/test_mcp_tools.py
index cc53a869..4d40b672 100644
--- a/backend/src/test/test_mcp_tools.py
+++ b/backend/src/test/test_mcp_tools.py
@@ -1,16 +1,13 @@
-"""Tests for module.mcp.tools - _bangumi_to_dict helper and _dispatch routing."""
+"""Tests for module.mcp.tools - _bangumi_to_dict helper and tool calls via AutoBangumiMCP."""
 
 import json
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
+from mcp import types
 
-from module.mcp.tools import (
-    TOOLS,
-    _bangumi_to_dict,
-    _dispatch,
-    handle_tool,
-)
+from module.mcp import mcp
+from module.mcp.tools import _bangumi_to_dict
 from module.models import Bangumi, ResponseModel
 from test.factories import make_bangumi
 
@@ -19,6 +16,18 @@ from test.factories import make_bangumi
 # ---------------------------------------------------------------------------
 
 
+def _json(result: types.CallToolResult):
+    """结果只有一段 JSON 文本，返回解析后的值。"""
+    [content] = result.content
+    assert isinstance(content, types.TextContent)
+    return json.loads(content.text)
+
+
+async def _call(name: str, args: dict):
+    """经 ``mcp.call_tool`` 调用工具，返回解析后的 JSON 文本。"""
+    return _json(await mcp.call_tool(name, args))
+
+
 def _make_response(status: bool = True, msg: str = "OK") -> ResponseModel:
     return ResponseModel(status=status, status_code=200, msg_en=msg, msg_zh=msg)
 
@@ -171,47 +180,56 @@ class TestBangumiToDict:
 
 
 # ---------------------------------------------------------------------------
-# TOOLS list
+# tool definitions
 # ---------------------------------------------------------------------------
 
 
 class TestToolsDefinitions:
-    """Sanity-check the static TOOLS list."""
+    """内置工具的名称与参数名必须与旧的手写 schema 一致，已有客户端才能继续使用。"""
 
-    def test_tools_is_list(self):
-        assert isinstance(TOOLS, list)
-
-    def test_tools_not_empty(self):
-        assert len(TOOLS) > 0
-
-    def test_all_tools_have_names(self):
-        for tool in TOOLS:
-            assert tool.name, f"Tool missing name: {tool}"
-
-    def test_expected_tool_names_present(self):
-        names = {t.name for t in TOOLS}
-        required = {
-            "list_anime",
-            "get_anime",
-            "search_anime",
-            "subscribe_anime",
-            "unsubscribe_anime",
-            "list_downloads",
-            "list_rss_feeds",
-            "get_program_status",
-            "refresh_feeds",
-            "update_anime",
+    async def test_tool_names_and_arguments_unchanged(self):
+        tools = {t.name: t for t in await mcp.list_tools()}
+        expected = {
+            "list_anime": {"active_only"},
+            "get_anime": {"id"},
+            "search_anime": {"keywords", "site"},
+            "subscribe_anime": {"rss_link", "parser"},
+            "unsubscribe_anime": {"id", "delete"},
+            "list_downloads": {"status"},
+            "list_rss_feeds": set(),
+            "get_program_status": set(),
+            "refresh_feeds": set(),
+            "update_anime": {
+                "id",
+                "episode_offset",
+                "season_offset",
+                "season",
+                "filter",
+            },
         }
-        assert required.issubset(names)
+        assert set(tools) == set(expected)
+        for name, args in expected.items():
+            assert set(tools[name].input_schema["properties"]) == args, name
+            assert tools[name].description
+            # 结果是 JSON 文本，不声明 outputSchema
+            assert tools[name].output_schema is None
+
+    async def test_required_and_enum_from_type_hints(self):
+        tools = {t.name: t for t in await mcp.list_tools()}
+        search = tools["search_anime"].input_schema
+        assert search["required"] == ["keywords"]
+        assert search["properties"]["site"]["enum"] == ["mikan", "dmhy", "nyaa"]
+        assert search["properties"]["site"]["default"] == "mikan"
+        assert tools["update_anime"].input_schema["required"] == ["id"]
 
 
 # ---------------------------------------------------------------------------
-# _dispatch routing
+# tool routing
 # ---------------------------------------------------------------------------
 
 
 class TestDispatch:
-    """Verify _dispatch delegates to the correct handler for each tool name."""
+    """Verify mcp.call_tool delegates to the correct handler for each tool name."""
 
     # --- list_anime ---
 
@@ -221,7 +239,7 @@ class TestDispatch:
         ctx = _mock_db(bangumi_list=bangumi)
 
         with patch("module.mcp.tools.Database", return_value=ctx):
-            result = await _dispatch("list_anime", {})
+            result = await _call("list_anime", {})
 
         assert isinstance(result, list)
         assert len(result) == 2
@@ -232,7 +250,7 @@ class TestDispatch:
         ctx, mock_mgr = _mock_sync_manager(bangumi_list=bangumi)
 
         with patch("module.mcp.tools.TorrentManager", return_value=ctx):
-            result = await _dispatch("list_anime", {"active_only": True})
+            result = await _call("list_anime", {"active_only": True})
 
         mock_mgr.search_all_bangumi.assert_called_once()
         assert len(result) == 1
@@ -245,7 +263,7 @@ class TestDispatch:
         ctx, _ = _mock_sync_manager(single=bangumi)
 
         with patch("module.mcp.tools.TorrentManager", return_value=ctx):
-            result = await _dispatch("get_anime", {"id": 5})
+            result = await _call("get_anime", {"id": 5})
 
         assert isinstance(result, dict)
         assert result["id"] == 5
@@ -262,7 +280,7 @@ class TestDispatch:
         ctx, _ = _mock_sync_manager(single=not_found)
 
         with patch("module.mcp.tools.TorrentManager", return_value=ctx):
-            result = await _dispatch("get_anime", {"id": 999})
+            result = await _call("get_anime", {"id": 999})
 
         assert isinstance(result, dict)
         assert "error" in result
@@ -285,7 +303,7 @@ class TestDispatch:
         mock_st.__aexit__ = AsyncMock(return_value=False)
 
         with patch("module.mcp.tools.SearchTorrent", return_value=mock_st):
-            result = await _dispatch(
+            result = await _call(
                 "search_anime", {"keywords": "One Piece", "site": "mikan"}
             )
 
@@ -308,7 +326,7 @@ class TestDispatch:
         mock_st.__aexit__ = AsyncMock(return_value=False)
 
         with patch("module.mcp.tools.SearchTorrent", return_value=mock_st):
-            await _dispatch("search_anime", {"keywords": "Bleach"})
+            await _call("search_anime", {"keywords": "Bleach"})
 
         assert captured_site == ["mikan"]
 
@@ -329,7 +347,7 @@ class TestDispatch:
                 new=AsyncMock(return_value=fake_resp),
             ),
         ):
-            result = await _dispatch(
+            result = await _call(
                 "subscribe_anime",
                 {"rss_link": "https://mikanani.me/RSS/test", "parser": "mikan"},
             )
@@ -348,7 +366,7 @@ class TestDispatch:
         mock_analyser.link_to_data = AsyncMock(return_value=fake_error)
 
         with patch("module.mcp.tools.RSSAnalyser", return_value=mock_analyser):
-            result = await _dispatch(
+            result = await _call(
                 "subscribe_anime",
                 {"rss_link": "https://bad-rss.example.com", "parser": "mikan"},
             )
@@ -363,7 +381,7 @@ class TestDispatch:
         mock_mgr.disable_rule = AsyncMock(return_value=_make_response(True, "Disabled"))
 
         with patch("module.mcp.tools.TorrentManager", return_value=ctx):
-            result = await _dispatch("unsubscribe_anime", {"id": 3, "delete": False})
+            result = await _call("unsubscribe_anime", {"id": 3, "delete": False})
 
         mock_mgr.disable_rule.assert_called_once_with(3)
         assert isinstance(result, dict)
@@ -375,7 +393,7 @@ class TestDispatch:
         mock_mgr.delete_rule = AsyncMock(return_value=_make_response(True, "Deleted"))
 
         with patch("module.mcp.tools.TorrentManager", return_value=ctx):
-            result = await _dispatch("unsubscribe_anime", {"id": 3, "delete": True})
+            result = await _call("unsubscribe_anime", {"id": 3, "delete": True})
 
         mock_mgr.delete_rule.assert_called_once_with(3)
         assert isinstance(result, dict)
@@ -404,7 +422,7 @@ class TestDispatch:
         with patch(
             "module.downloader.download_client.DownloadClient", return_value=mock_client
         ):
-            result = await _dispatch("list_downloads", {"status": "all"})
+            result = await _call("list_downloads", {"status": "all"})
 
         mock_client.get_torrent_info.assert_called_once_with(
             status_filter=None, category="Bangumi"
@@ -422,7 +440,7 @@ class TestDispatch:
         with patch(
             "module.downloader.download_client.DownloadClient", return_value=mock_client
         ):
-            await _dispatch("list_downloads", {"status": "downloading"})
+            await _call("list_downloads", {"status": "downloading"})
 
         mock_client.get_torrent_info.assert_called_once_with(
             status_filter="downloading", category="Bangumi"
@@ -450,7 +468,7 @@ class TestDispatch:
         with patch(
             "module.downloader.download_client.DownloadClient", return_value=mock_client
         ):
-            result = await _dispatch("list_downloads", {})
+            result = await _call("list_downloads", {})
 
         expected_keys = {
             "name",
@@ -482,7 +500,7 @@ class TestDispatch:
         ctx = _mock_db(feeds=[fake_feed])
 
         with patch("module.mcp.tools.Database", return_value=ctx):
-            result = await _dispatch("list_rss_feeds", {})
+            result = await _call("list_rss_feeds", {})
 
         assert isinstance(result, list)
         assert result[0]["name"] == "Mikan"
@@ -498,18 +516,11 @@ class TestDispatch:
 
         with (
             patch("module.mcp.tools.VERSION", "3.2.0-beta"),
-            patch("module.mcp.tools._get_program_status") as mock_fn,
+            patch("module.mcp.tools.get_context", return_value=mock_program),
         ):
-            mock_fn.return_value = {
-                "version": "3.2.0-beta",
-                "running": True,
-                "first_run": False,
-            }
-            result = await _dispatch("get_program_status", {})
+            result = await _call("get_program_status", {})
 
-        assert "version" in result
-        assert "running" in result
-        assert "first_run" in result
+        assert result == {"version": "3.2.0-beta", "running": True, "first_run": False}
 
     # --- refresh_feeds ---
 
@@ -526,7 +537,7 @@ class TestDispatch:
             patch("module.mcp.tools.DownloaderPool", return_value=mock_client),
             patch("module.mcp.tools.RSSEngine", return_value=mock_engine),
         ):
-            result = await _dispatch("refresh_feeds", {})
+            result = await _call("refresh_feeds", {})
 
         assert isinstance(result, dict)
         assert result["status"] is True
@@ -547,7 +558,7 @@ class TestDispatch:
             patch("module.mcp.tools.Database", return_value=db_ctx),
             patch("module.mcp.tools.TorrentManager", return_value=ctx),
         ):
-            result = await _dispatch(
+            result = await _call(
                 "update_anime",
                 {"id": 7, "episode_offset": 12, "season": 2},
             )
@@ -562,7 +573,7 @@ class TestDispatch:
         mock_mgr.bangumi.search_id.return_value = None
 
         with patch("module.mcp.tools.TorrentManager", return_value=ctx):
-            result = await _dispatch("update_anime", {"id": 9999})
+            result = await _call("update_anime", {"id": 9999})
 
         assert isinstance(result, dict)
         assert "error" in result
@@ -571,60 +582,44 @@ class TestDispatch:
     # --- unknown tool ---
 
     async def test_dispatch_unknown_tool(self):
-        """An unrecognised tool name returns an error dict."""
-        result = await _dispatch("does_not_exist", {})
-        assert isinstance(result, dict)
-        assert "error" in result
-        assert "does_not_exist" in result["error"]
+        """An unrecognised tool name returns an is_error result."""
+        result = await mcp.call_tool("does_not_exist", {})
+        assert result.is_error
+        body = _json(result)
+        assert body == {"error": "Unknown tool: does_not_exist"}
 
 
 # ---------------------------------------------------------------------------
-# handle_tool wrapper
+# CallToolResult
 # ---------------------------------------------------------------------------
 
 
-class TestHandleTool:
-    """Verify handle_tool wraps results correctly and handles exceptions."""
+class TestCallToolResult:
+    """Verify call_tool wraps results in one JSON text block and turns errors into is_error."""
 
-    async def test_handle_tool_returns_text_content_list(self):
-        """handle_tool always returns a list of TextContent objects."""
-        from mcp import types
-
-        bangumi = [make_bangumi(id=1)]
+    async def test_result_is_single_json_text_block(self):
+        """列表结果也只有一段 JSON 文本（MCPServer 默认会拆成每项一段）。"""
+        bangumi = [make_bangumi(id=1), make_bangumi(id=2, title_raw="Another")]
         ctx = _mock_db(bangumi_list=bangumi)
 
         with patch("module.mcp.tools.Database", return_value=ctx):
-            result = await handle_tool("list_anime", {})
+            result = await mcp.call_tool("list_anime", {})
 
-        assert isinstance(result, list)
-        assert all(isinstance(item, types.TextContent) for item in result)
+        assert not result.is_error
+        assert [b["id"] for b in _json(result)] == [1, 2]
 
-    async def test_handle_tool_result_is_valid_json(self):
-        """The text in TextContent is valid JSON."""
-        bangumi = [make_bangumi(id=1)]
-        ctx = _mock_db(bangumi_list=bangumi)
-
-        with patch("module.mcp.tools.Database", return_value=ctx):
-            result = await handle_tool("list_anime", {})
-
-        parsed = json.loads(result[0].text)
-        assert isinstance(parsed, list)
-
-    async def test_handle_tool_exception_returns_error_json(self):
-        """If the underlying handler raises, handle_tool returns a JSON error."""
+    async def test_exception_returns_is_error_with_message(self):
+        """处理函数抛出异常时返回 is_error 结果，错误文本保留原始消息。"""
         with patch(
-            "module.mcp.tools._dispatch",
-            new=AsyncMock(side_effect=RuntimeError("something broke")),
+            "module.mcp.tools.Database",
+            side_effect=RuntimeError("something broke"),
         ):
-            result = await handle_tool("list_anime", {})
+            result = await mcp.call_tool("list_anime", {})
 
-        assert len(result) == 1
-        body = json.loads(result[0].text)
-        assert "error" in body
-        assert "something broke" in body["error"]
+        assert result.is_error
+        assert _json(result) == {"error": "something broke"}
 
-    async def test_handle_tool_unknown_name_returns_error_json(self):
-        """An unknown tool name bubbles up as a JSON error via handle_tool."""
-        result = await handle_tool("totally_unknown_tool", {})
-        body = json.loads(result[0].text)
-        assert "error" in body
+    async def test_invalid_arguments_return_is_error(self):
+        result = await mcp.call_tool("get_anime", {"id": "abc"})
+        assert result.is_error
+        assert "id" in _json(result)["error"]
diff --git a/backend/src/test/test_plugin_events_api.py b/backend/src/test/test_plugin_events_api.py
index 8fa6d287..62954ca3 100644
--- a/backend/src/test/test_plugin_events_api.py
+++ b/backend/src/test/test_plugin_events_api.py
@@ -4,6 +4,7 @@ import json
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import mcp.types as mcp_types
 import pytest
 from fastapi import APIRouter, FastAPI, HTTPException
 from fastapi.testclient import TestClient
@@ -383,9 +384,29 @@ async def _search(args):
     return {"q": args.get("q"), "hits": [1, 2]}
 
 
+def _tool_json(result: mcp_types.CallToolResult):
+    """工具结果只有一段 JSON 文本，返回解析后的值。"""
+    [content] = result.content
+    assert isinstance(content, mcp_types.TextContent)
+    return json.loads(content.text)
+
+
+async def _read_text(uri: str) -> str:
+    from mcp.server.lowlevel.helper_types import ReadResourceContents
+
+    from module.mcp import mcp
+
+    [content] = await mcp.read_resource(uri)
+    assert isinstance(content, ReadResourceContents)
+    assert isinstance(content.content, str)
+    return content.content
+
+
 class TestMcp:
-    def test_plugin_tools_listed_with_namespace(self, registry):
-        from module.mcp.tools import TOOLS, all_tools
+    async def test_plugin_tools_listed_with_namespace(self, registry):
+        from mcp.server.mcpserver import MCPServer
+
+        from module.mcp import mcp
 
         add(
             registry,
@@ -394,27 +415,29 @@ class TestMcp:
             McpTool("Search", _search, {"type": "object"}),
             plugin_id="demo",
         )
-        names = [t.name for t in all_tools()]
-        assert names[: len(TOOLS)] == [t.name for t in TOOLS]
-        assert "demo__search" in names
+        builtin = [t.name for t in await MCPServer.list_tools(mcp)]
+        names = [t.name for t in await mcp.list_tools()]
+        assert names == builtin + ["demo__search"]
 
     async def test_call_dispatches_to_plugin(self, registry):
-        from module.mcp.tools import handle_tool
+        from module.mcp import mcp
 
         add(registry, points.MCP_TOOL, "search", McpTool("S", _search), "demo")
-        [content] = await handle_tool("demo__search", {"q": "x"})
-        assert json.loads(content.text) == {"q": "x", "hits": [1, 2]}
+        result = await mcp.call_tool("demo__search", {"q": "x"})
+        assert not result.is_error
+        assert _tool_json(result) == {"q": "x", "hits": [1, 2]}
 
     async def test_unknown_names_keep_error(self, registry):
-        from module.mcp.tools import handle_tool
+        from module.mcp import mcp
 
         add(registry, points.MCP_TOOL, "search", McpTool("S", _search), "demo")
         for name in ("nope", "demo__nope", "other__search"):
-            [content] = await handle_tool(name, {})
-            assert json.loads(content.text) == {"error": f"Unknown tool: {name}"}
+            result = await mcp.call_tool(name, {})
+            assert result.is_error
+            assert _tool_json(result) == {"error": f"Unknown tool: {name}"}
 
     async def test_failure_returns_error_and_counts(self, registry, breaker):
-        from module.mcp.tools import handle_tool
+        from module.mcp import mcp
 
         async def boom(args):
             raise RuntimeError("plugin bug")
@@ -422,10 +445,29 @@ class TestMcp:
         add(registry, points.MCP_TOOL, "boom", McpTool("B", boom), "demo")
         ctx = SimpleNamespace(plugins=SimpleNamespace(breaker=breaker))
         with patch("module.mcp.plugins.get_context", return_value=ctx):
-            [content] = await handle_tool("demo__boom", {})
-        assert json.loads(content.text) == {"error": "plugin bug"}
+            result = await mcp.call_tool("demo__boom", {})
+        assert result.is_error
+        assert _tool_json(result) == {"error": "plugin bug"}
         assert breaker._failures["demo"] == 1
 
+    async def test_invalid_arguments_rejected_before_handler(self, registry, breaker):
+        from module.mcp import mcp
+
+        handler = AsyncMock()
+        schema = {
+            "type": "object",
+            "properties": {"q": {"type": "string"}},
+            "required": ["q"],
+        }
+        add(registry, points.MCP_TOOL, "search", McpTool("S", handler, schema), "demo")
+        ctx = SimpleNamespace(plugins=SimpleNamespace(breaker=breaker))
+        with patch("module.mcp.plugins.get_context", return_value=ctx):
+            result = await mcp.call_tool("demo__search", {})
+        assert result.is_error
+        assert "'q' is a required property" in _tool_json(result)["error"]
+        handler.assert_not_called()
+        assert "demo" not in breaker._failures
+
     def test_invalid_names_and_bad_factories_skipped(self, registry):
         from module.mcp.plugins import list_plugin_tools
 
@@ -435,7 +477,10 @@ class TestMcp:
         assert [t.name for t in list_plugin_tools()] == ["demo__ok"]
 
     async def test_plugin_resources(self, registry):
-        from module.mcp.resources import RESOURCES, all_resources, handle_resource
+        from mcp.server.mcpserver import MCPServer
+        from mcp.server.mcpserver.exceptions import ResourceNotFoundError
+
+        from module.mcp import mcp
 
         async def stats():
             return {"count": 3}
@@ -445,24 +490,26 @@ class TestMcp:
 
         add(registry, points.MCP_RESOURCE, "stats", McpResource("Stats", stats), "demo")
         add(registry, points.MCP_RESOURCE, "text", McpResource("Text", text), "demo")
-        uris = [str(r.uri) for r in all_resources()]
-        assert uris[: len(RESOURCES)] == [str(r.uri) for r in RESOURCES]
+        builtin = [r.uri for r in await MCPServer.list_resources(mcp)]
+        uris = [r.uri for r in await mcp.list_resources()]
+        assert uris[: len(builtin)] == builtin
         assert "autobangumi://plugins/demo/stats" in uris
-        stats_json = await handle_resource("autobangumi://plugins/demo/stats")
+        stats_json = await _read_text("autobangumi://plugins/demo/stats")
         assert json.loads(stats_json) == {"count": 3}
-        assert await handle_resource("autobangumi://plugins/demo/text") == "plain"
-        missing = await handle_resource("autobangumi://plugins/demo/none")
-        assert json.loads(missing) == {
-            "error": "Unknown resource: autobangumi://plugins/demo/none"
-        }
+        assert await _read_text("autobangumi://plugins/demo/text") == "plain"
+        with pytest.raises(
+            ResourceNotFoundError,
+            match="Unknown resource: autobangumi://plugins/demo/none",
+        ):
+            await mcp.read_resource("autobangumi://plugins/demo/none")
 
     @pytest.mark.parametrize("resource_id", ["stats", "Weekly Stats", "统计"])
     async def test_read_resource_via_protocol_returns_plugin_content(
         self, registry, resource_id
     ):
-        from mcp import types
+        from mcp import Client, types
 
-        from module.mcp import server
+        from module.mcp import mcp
 
         async def stats():
             return {"count": 3}
@@ -474,20 +521,14 @@ class TestMcp:
             McpResource("Stats", stats),
             "demo",
         )
-        [listed] = [
-            r
-            for r in await server.list_resources()
-            if str(r.uri).startswith("autobangumi://plugins/")
-        ]
-        handler = server.server.request_handlers[types.ReadResourceRequest]
-        result = await handler(
-            types.ReadResourceRequest(
-                method="resources/read",
-                params=types.ReadResourceRequestParams(uri=listed.uri),
-            )
-        )
-        assert isinstance(result.root, types.ReadResourceResult)
-        [content] = result.root.contents
+        async with Client(mcp) as client:
+            [listed] = [
+                r
+                for r in (await client.list_resources()).resources
+                if r.uri.startswith("autobangumi://plugins/")
+            ]
+            result = await client.read_resource(listed.uri)
+        [content] = result.contents
         assert isinstance(content, types.TextResourceContents)
         assert json.loads(content.text) == {"count": 3}
 
@@ -498,17 +539,18 @@ class TestMcp:
             (points.MCP_RESOURCE, McpResource(None, _search)),  # type: ignore[arg-type]
         ],
     )
-    def test_list_malformed_plugin_spec_skipped_and_counted(
+    async def test_list_malformed_plugin_spec_skipped_and_counted(
         self, registry, breaker, point, impl
     ):
-        from module.mcp.resources import RESOURCES, all_resources
-        from module.mcp.tools import TOOLS, all_tools
+        from mcp.server.mcpserver import MCPServer
+
+        from module.mcp import mcp
 
         add(registry, point, "bad", impl, "demo")
         ctx = SimpleNamespace(plugins=SimpleNamespace(breaker=breaker))
         with patch("module.mcp.plugins.get_context", return_value=ctx):
-            assert len(all_tools()) == len(TOOLS)
-            assert len(all_resources()) == len(RESOURCES)
+            assert await mcp.list_tools() == await MCPServer.list_tools(mcp)
+            assert await mcp.list_resources() == await MCPServer.list_resources(mcp)
         assert breaker._failures["demo"] == 1
 
     async def test_read_plugin_resource_hung_handler_times_out_and_counts(
@@ -516,8 +558,9 @@ class TestMcp:
     ):
         import asyncio
 
-        from module.mcp import plugins
-        from module.mcp.resources import handle_resource
+        from mcp.server.mcpserver.exceptions import ResourceError
+
+        from module.mcp import mcp, plugins
 
         async def hang():
             await asyncio.Event().wait()
@@ -525,16 +568,25 @@ class TestMcp:
         monkeypatch.setattr(plugins, "DEFAULT_TOOL_TIMEOUT", 0.01)
         add(registry, points.MCP_RESOURCE, "hang", McpResource("H", hang), "demo")
         ctx = SimpleNamespace(plugins=SimpleNamespace(breaker=breaker))
-        with patch("module.mcp.plugins.get_context", return_value=ctx):
-            raw = await handle_resource("autobangumi://plugins/demo/hang")
-        assert "error" in json.loads(raw)
+        with (
+            patch("module.mcp.plugins.get_context", return_value=ctx),
+            pytest.raises(ResourceError, match="TimeoutError"),
+        ):
+            await mcp.read_resource("autobangumi://plugins/demo/hang")
         assert breaker._failures["demo"] == 1
 
-    async def test_mcp_server_handlers_include_plugins(self, registry):
-        from module.mcp import server
+    async def test_plugin_tool_via_protocol(self, registry):
+        from mcp import Client
+
+        from module.mcp import mcp
 
         add(registry, points.MCP_TOOL, "search", McpTool("S", _search), "demo")
-        assert "demo__search" in [t.name for t in await server.list_tools()]
+        async with Client(mcp) as client:
+            listed = await client.list_tools()
+            assert "demo__search" in [t.name for t in listed.tools]
+            result = await client.call_tool("demo__search", {"q": "x"})
+        assert not result.is_error
+        assert _tool_json(result) == {"q": "x", "hits": [1, 2]}
 
 
 # ---------------------------------------------------------------- end to end
@@ -589,7 +641,7 @@ entry = "x:Y"
 async def test_plugin_end_to_end(registry, monkeypatch, tmp_path):
     from module.core import AppContext
     from module.core.scheduler import Scheduler
-    from module.mcp.tools import all_tools, handle_tool
+    from module.mcp import mcp
 
     # AppContext 会设置进程级 bus/runner；用 monkeypatch 让测试结束后复原
     monkeypatch.setattr(host, "_bus", None)
@@ -632,9 +684,9 @@ async def test_plugin_end_to_end(registry, monkeypatch, tmp_path):
         assert title == f"yo: {event.describe()[0]}"
 
         # MCP
-        assert "full__greet" in [t.name for t in all_tools()]
-        [content] = await handle_tool("full__greet", {})
-        assert json.loads(content.text) == {"greeting": "yo"}
+        assert "full__greet" in [t.name for t in await mcp.list_tools()]
+        result = await mcp.call_tool("full__greet", {})
+        assert _tool_json(result) == {"greeting": "yo"}
 
         # 路由
         app = FastAPI()
@@ -652,44 +704,7 @@ async def test_plugin_end_to_end(registry, monkeypatch, tmp_path):
         settings_obj.plugins.enabled["full"] = False
         await manager.apply_settings()
         assert client.get("/api/v1/plugins/full/greet").status_code == 404
-        assert "full__greet" not in [t.name for t in all_tools()]
+        assert "full__greet" not in [t.name for t in await mcp.list_tools()]
         assert await render_event(event, "bark") == event.describe()
     finally:
         await manager.stop()
-
-
-# ---------------------------------------------------------------- MCP SSE
-
-
-async def test_mcp_sse_endpoint_sends_response_only_once(monkeypatch):
-    """/sse 结束后不能再发第二次 http.response.start（3.x 起每次断开都报
-    AssertionError）。"""
-    from contextlib import asynccontextmanager
-
-    from module.mcp import server as mcp_server
-
-    @asynccontextmanager
-    async def fake_connect_sse(scope, receive, send):
-        await send({"type": "http.response.start", "status": 200, "headers": []})
-        await send({"type": "http.response.body", "body": b"", "more_body": False})
-        yield (object(), object())
-
-    async def fake_run(*args, **kwargs):
-        return None
-
-    monkeypatch.setattr(mcp_server.sse, "connect_sse", fake_connect_sse)
-    monkeypatch.setattr(mcp_server.server, "run", fake_run)
-    sent: list[dict] = []
-
-    async def receive():
-        return {"type": "http.disconnect"}
-
-    async def send(message):
-        sent.append(message)
-
-    await mcp_server.handle_sse(
-        {"type": "http", "method": "GET", "path": "/sse", "headers": []},
-        receive,
-        send,
-    )
-    assert [m["type"] for m in sent].count("http.response.start") == 1
diff --git a/backend/src/test/test_plugin_web.py b/backend/src/test/test_plugin_web.py
index f072418d..b067c3b0 100644
--- a/backend/src/test/test_plugin_web.py
+++ b/backend/src/test/test_plugin_web.py
@@ -200,8 +200,8 @@ class TestSpaCsp:
         assert resp.status_code == 200
         assert "content-security-policy" not in resp.headers
 
-    def test_mcp_mount_wins_over_fallback(self, spa):
-        resp = spa.get("/mcp/sse", headers=HTML)
+    def test_mcp_route_wins_over_fallback(self, spa):
+        resp = spa.get("/mcp", headers=HTML)
         assert resp.status_code != 200
         assert "content-security-policy" not in resp.headers
 
diff --git a/backend/uv.lock b/backend/uv.lock
index 5ed82ee4..fae6a391 100644
--- a/backend/uv.lock
+++ b/backend/uv.lock
@@ -150,6 +150,7 @@ dependencies = [
     { name = "google-genai" },
     { name = "httpx2", extra = ["socks"] },
     { name = "jinja2" },
+    { name = "jsonschema" },
     { name = "mcp" },
     { name = "openai" },
     { name = "packaging" },
@@ -184,7 +185,8 @@ requires-dist = [
     { name = "google-genai", specifier = ">=2.10.0" },
     { name = "httpx2", extras = ["socks"], specifier = ">=2.15" },
     { name = "jinja2", specifier = ">=3.1.2" },
-    { name = "mcp", specifier = ">=1.8.0" },
+    { name = "jsonschema", specifier = ">=4.20.0" },
+    { name = "mcp", specifier = ">=2.3.0" },
     { name = "openai", specifier = ">=1.54.3" },
     { name = "packaging", specifier = ">=26.3" },
     { name = "pydantic", specifier = ">=2.14.1" },
@@ -620,15 +622,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/2a/39/e50c7c3a983047577ee07d2a9e53faf5a69493943ec3f6a384bdc792deb2/httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad", size = 73517, upload-time = "2024-12-06T15:37:21.509Z" },
 ]
 
-[[package]]
-name = "httpx-sse"
-version = "0.4.3"
-source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/0f/4c/751061ffa58615a32c31b2d82e8482be8dd4a89154f003147acee90f2be9/httpx_sse-0.4.3.tar.gz", hash = "sha256:9b1ed0127459a66014aec3c56bebd93da3c1bc8bb6618c8082039a44889a755d", size = 15943, upload-time = "2025-10-10T21:48:22.271Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/d2/fd/6668e5aec43ab844de6fc74927e155a3b37bf40d7c3790e49fc0406b6578/httpx_sse-0.4.3-py3-none-any.whl", hash = "sha256:0ac1c9fe3c0afad2e0ebb25a934a59f4c7823b60792691f779fad2c5568830fc", size = 8960, upload-time = "2025-10-10T21:48:21.158Z" },
-]
-
 [[package]]
 name = "httpx2"
 version = "2.15.1"
@@ -881,15 +874,15 @@ wheels = [
 
 [[package]]
 name = "mcp"
-version = "1.26.0"
+version = "2.3.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "anyio" },
-    { name = "httpx" },
-    { name = "httpx-sse" },
+    { name = "httpx2" },
     { name = "jsonschema" },
+    { name = "mcp-types" },
+    { name = "opentelemetry-api" },
     { name = "pydantic" },
-    { name = "pydantic-settings" },
     { name = "pyjwt", extra = ["crypto"] },
     { name = "python-multipart" },
     { name = "pywin32", marker = "sys_platform == 'win32'" },
@@ -899,9 +892,22 @@ dependencies = [
     { name = "typing-inspection" },
     { name = "uvicorn", marker = "sys_platform != 'emscripten'" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/fc/6d/62e76bbb8144d6ed86e202b5edd8a4cb631e7c8130f3f4893c3f90262b10/mcp-1.26.0.tar.gz", hash = "sha256:db6e2ef491eecc1a0d93711a76f28dec2e05999f93afd48795da1c1137142c66", size = 608005, upload-time = "2026-01-24T19:40:32.468Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/9d/8d/e0d339616f4810e9051d4aba6887afab289ab1f81875fe908b606cdfd0e3/mcp-2.3.0.tar.gz", hash = "sha256:8b147a50441cf059dc88c684e0aeed3687f0aa0f39c6cde7b90330effd2b34d8", size = 4171955, upload-time = "2026-10-02T22:06:56.091Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/fd/d9/eaa1f80170d2b7c5ba23f3b59f766f3a0bb41155fbc32a69adfa1adaaef9/mcp-1.26.0-py3-none-any.whl", hash = "sha256:904a21c33c25aa98ddbeb47273033c435e595bbacfdb177f4bd87f6dceebe1ca", size = 233615, upload-time = "2026-01-24T19:40:30.652Z" },
+    { url = "https://files.pythonhosted.org/packages/20/d7/4a2878f128df729db3ca04e71bfce26405cc2e0c9ebe19c592feeb4c61b9/mcp-2.3.0-py3-none-any.whl", hash = "sha256:dd0c44c089d16453e8ae31a3877a0054d7a2314caaa81f5e0541b9b1734b2377", size = 368664, upload-time = "2026-10-02T22:06:52.119Z" },
+]
+
+[[package]]
+name = "mcp-types"
+version = "2.3.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "pydantic" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/9e/2d/7c251e34207f6c51000312fc8839111ac45cfe02023f90b44e7f1051dd8e/mcp_types-2.3.0.tar.gz", hash = "sha256:d1e46549edb35ee19a94940fcee6d1addd7e589ab7ea92dda83f5d84781fc362", size = 65898, upload-time = "2026-10-02T22:06:57.863Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/ce/48/32583066b155bb5d4507e7887dc88ee23118cc4de8e2a12870c49719e2a8/mcp_types-2.3.0-py3-none-any.whl", hash = "sha256:968efdbdaedfab06adae40d378a34395f1090c5921d4be3c9cde283aaf76d91d", size = 69108, upload-time = "2026-10-02T22:06:54.099Z" },
 ]
 
 [[package]]
@@ -1128,20 +1134,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/6d/56/8a702c27e5be9f47e5f19d8669227424290e4c024e7c370279cbaf244b4e/pydantic_core-2.50.1-cp315-cp315t-win_arm64.whl", hash = "sha256:c3ede305158e75510be50869b319550ab072008c13d64d4ab1e094fb286b6f44", size = 2175297, upload-time = "2026-10-11T18:35:04.079Z" },
 ]
 
-[[package]]
-name = "pydantic-settings"
-version = "2.13.1"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "pydantic" },
-    { name = "python-dotenv" },
-    { name = "typing-inspection" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/52/6d/fffca34caecc4a3f97bda81b2098da5e8ab7efc9a66e819074a11955d87e/pydantic_settings-2.13.1.tar.gz", hash = "sha256:b4c11847b15237fb0171e1462bf540e294affb9b86db4d9aa5c01730bdbe4025", size = 223826, upload-time = "2026-02-19T13:45:08.055Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/00/4b/ccc026168948fec4f7555b9164c724cf4125eac006e176541483d2c959be/pydantic_settings-2.13.1-py3-none-any.whl", hash = "sha256:d56fd801823dbeae7f0975e1f8c8e25c258eb75d278ea7abb5d9cebb01b56237", size = 58929, upload-time = "2026-02-19T13:45:06.034Z" },
-]
-
 [[package]]
 name = "pygments"
 version = "2.19.2"
diff --git a/docs/api/index.md b/docs/api/index.md
index 2515fbcf..dc74974d 100644
--- a/docs/api/index.md
+++ b/docs/api/index.md
@@ -208,7 +208,7 @@ curl -N -H "Authorization: Bearer $AB_TOKEN" http://your-host:7892/api/v1/events
 
 ## MCP
 
-AB 在根路径的 `/mcp` 下提供 MCP 服务（SSE 传输）：客户端连接 `GET /mcp/sse`，并向 `POST /mcp/messages/` 发送消息。
+AB 在根路径的 `/mcp` 提供 MCP 服务（Streamable HTTP 传输）：在 MCP 客户端中把服务地址设为 `http://<AB 地址>:7892/mcp`。3.x 与 4.0.0-beta.2 及更早版本的 SSE 端点 `/mcp/sse` 已删除。
 
 - 访问控制与 REST API 分开：客户端 IP 在 `security.mcp_whitelist` 内，或请求带 `scope=mcp` 的令牌（`Authorization: Bearer <令牌>`）。`mcp_whitelist` 为空时拒绝所有基于 IP 的访问，令牌仍然有效。
 - 内置工具：`list_anime`、`get_anime`、`search_anime`、`subscribe_anime`、`unsubscribe_anime`、`list_downloads`、`list_rss_feeds`、`get_program_status`、`refresh_feeds`、`update_anime`。
diff --git a/docs/changelog/4.0.md b/docs/changelog/4.0.md
index 099617f8..98963cc5 100644
--- a/docs/changelog/4.0.md
+++ b/docs/changelog/4.0.md
@@ -4,6 +4,16 @@
 
 升级步骤、回滚方法与升级后的检查项见 [从 3.3 升级到 4.0](../deploy/upgrade-4.0)。
 
+## 4.0.0-beta.3
+
+### 破坏性变更
+
+- MCP 改用 Streamable HTTP，端点为 `/mcp`，旧的 `/mcp/sse` 已删除，请更新 MCP 客户端的地址。
+- 带 JSON 请求体的 API 调用必须发送 `Content-Type: application/json`，否则返回 422。
+- 未知的 `/api/…` 路径返回 404，不再返回 WebUI 页面。
+- 插件 SDK（`ab_sdk`）0.6.0：`AdapterContext.build_http_client` 返回 `httpx2.AsyncClient`，只调用 `get` / `post` 的插件不需要改动，直接使用 httpx 专有 API 的插件必须更新。
+- 运行环境：Python 3.14。
+
 ## 4.0.0-beta.1
 
 Docker 镜像：`ghcr.io/estrellaxd/auto_bangumi:4.0.0-beta.1`（`dev-latest` 也指向最新测试版）。
diff --git a/docs/en/api/index.md b/docs/en/api/index.md
index 8058da7e..140ff180 100644
--- a/docs/en/api/index.md
+++ b/docs/en/api/index.md
@@ -208,7 +208,7 @@ These endpoints need no authentication. `GET /setup/status` is always available
 
 ## MCP
 
-AB provides an MCP server under `/mcp` at the root path (SSE transport). A client connects to `GET /mcp/sse` and sends messages to `POST /mcp/messages/`.
+AB provides an MCP server at `/mcp` at the root path (Streamable HTTP transport). In the MCP client, set the server URL to `http://<AB address>:7892/mcp`. The SSE endpoint `/mcp/sse` of 3.x and 4.0.0-beta.2 and earlier is removed.
 
 - Access control is different from the REST API. The client IP must be in `security.mcp_whitelist`, or the request must have a token with `scope=mcp` (`Authorization: Bearer <token>`). If `mcp_whitelist` is empty, AB refuses all IP-based access. Tokens still work.
 - Built-in tools: `list_anime`, `get_anime`, `search_anime`, `subscribe_anime`, `unsubscribe_anime`, `list_downloads`, `list_rss_feeds`, `get_program_status`, `refresh_feeds`, `update_anime`.
diff --git a/docs/ja/api/index.md b/docs/ja/api/index.md
index b5208c6b..dbc76003 100644
--- a/docs/ja/api/index.md
+++ b/docs/ja/api/index.md
@@ -208,7 +208,7 @@ curl -N -H "Authorization: Bearer $AB_TOKEN" http://your-host:7892/api/v1/events
 
 ## MCP
 
-AB はルートパスの `/mcp` 以下で MCP サーバー（SSE トランスポート）を提供します。クライアントは `GET /mcp/sse` に接続し、`POST /mcp/messages/` にメッセージを送ります。
+AB はルートパスの `/mcp` で MCP サーバー（Streamable HTTP トランスポート）を提供します。MCP クライアントにはサーバーの URL として `http://<AB のアドレス>:7892/mcp` を設定します。3.x と 4.0.0-beta.2 以前の SSE エンドポイント `/mcp/sse` は削除されました。
 
 - アクセス制御は REST API とは別です。クライアントの IP が `security.mcp_whitelist` に含まれるか、リクエストに `scope=mcp` のトークン（`Authorization: Bearer <トークン>`）が必要です。`mcp_whitelist` が空の場合、IP によるアクセスはすべて拒否されます。トークンは引き続き使えます。
 - 組み込みツール：`list_anime`、`get_anime`、`search_anime`、`subscribe_anime`、`unsubscribe_anime`、`list_downloads`、`list_rss_feeds`、`get_program_status`、`refresh_feeds`、`update_anime`。
diff --git a/webui/e2e/specs/session-access.spec.ts b/webui/e2e/specs/session-access.spec.ts
index 4bbed1f0..f0cdc697 100644
--- a/webui/e2e/specs/session-access.spec.ts
+++ b/webui/e2e/specs/session-access.spec.ts
@@ -106,7 +106,7 @@ test('API and MCP tokens are isolated and Bearer cannot control accounts', async
     await bearerApi.bearerGet('/api/v1/users', apiToken.token, 403);
     await bearerApi.bearerGet('/api/v1/tokens', apiToken.token, 403);
     await bearerApi.bearerGet('/api/v1/bangumi/get/all', mcpToken.token, 401);
-    await bearerApi.bearerGet('/mcp/sse', apiToken.token, 403);
+    await bearerApi.bearerGet('/mcp', apiToken.token, 403);
 
     // An explicit invalid Authorization header must beat the valid admin cookie.
     await api.bearerGet('/api/v1/bangumi/get/all', mcpToken.token, 401);
@@ -114,8 +114,23 @@ test('API and MCP tokens are isolated and Bearer cannot control accounts', async
     const controller = new AbortController();
     const timeout = setTimeout(() => controller.abort(), 5000);
     try {
-      const response = await fetch(`${environment.baseURL}/mcp/sse`, {
-        headers: { Authorization: `Bearer ${mcpToken.token}` },
+      const response = await fetch(`${environment.baseURL}/mcp`, {
+        method: 'POST',
+        headers: {
+          Authorization: `Bearer ${mcpToken.token}`,
+          Accept: 'application/json, text/event-stream',
+          'Content-Type': 'application/json',
+        },
+        body: JSON.stringify({
+          jsonrpc: '2.0',
+          id: 1,
+          method: 'initialize',
+          params: {
+            protocolVersion: '2025-06-18',
+            capabilities: {},
+            clientInfo: { name: 'e2e', version: '0' },
+          },
+        }),
         signal: controller.signal,
       });
       expect(response.status).toBe(200);