feat(mcp): add factory domain tools

apps/bi/services.py

@@ -1,11 +1,15 @@
-from rest_framework.exceptions import ParseError
+import concurrent.futures
 import json
-from jinja2 import Template
+import logging
+
+from rest_framework.exceptions import ParseError
+
 from apps.bi.models import Dataset
-import concurrent
 from apps.utils.sql import execute_raw_sql, format_sqldata
 from apps.utils.tools import MyJSONEncoder
 
+myLogger = logging.getLogger('log')
+
 forbidden_keywords = ["UPDATE", "DELETE", "DROP", "TRUNCATE", "INSERT", "CREATE", "ALTER", "GRANT", "REVOKE", "EXEC", "EXECUTE"]
 
 
@@ -32,31 +36,57 @@ def format_json_with_placeholders(json_str, **kwargs):
     return formatted_json
 
 
-def exec_dataset(dt: Dataset, xquery: dict = {}):
+def render_dataset_sql(dt: Dataset, xquery=None, *, is_test=False):
+    """根据数据集配置和调用参数生成经过安全检查的只读 SQL。"""
+    query = dict(dt.default_param or {})
+    query.update(dict(dt.test_param or {}) if is_test else dict(xquery or {}))
+    if not dt.sql_query:
+        return ''
+    try:
+        return check_sql_safe(dt.sql_query.format(**query))
+    except KeyError as exc:
+        raise ParseError(f'需指定查询参数_{str(exc)}') from exc
+
+
+def execute_rendered_dataset(dt: Dataset, full_sql: str, *, raise_exception=True):
+    """执行已经渲染和校验的 SQL，返回可合并到数据集响应的结果。"""
+    results = {}
+    results2 = {}
+    can_cache = True
+    sql_list = [sql for sql in full_sql.strip(';').split(';') if sql.strip()]
+    if sql_list:
+        with concurrent.futures.ThreadPoolExecutor(max_workers=6) as executor:
+            futures = {
+                executor.submit(execute_raw_sql, sql): (f'ds{index}', sql)
+                for index, sql in enumerate(sql_list)
+            }
+            for future in concurrent.futures.as_completed(futures):
+                name, sql = futures[future]
+                try:
+                    res = future.result()
+                    results[name], results2[name] = format_sqldata(res[0], res[1])
+                except Exception as exc:
+                    myLogger.error(f'bi查询异常：{str(exc)}-{dt.code}--{sql}')
+                    if raise_exception:
+                        raise ParseError(f'查询异常：{str(exc)}') from exc
+                    results[name] = 'error: ' + str(exc)
+                    can_cache = False
+
+    response_data = {'data': results, 'data2': results2}
+    if dt.echart_options and not dt.echart_options.startswith('function'):
+        for result in results.values():
+            if isinstance(result, str):
+                raise ParseError(result)
+        response_data['echart_options'] = format_json_with_placeholders(
+            dt.echart_options, **results
+        )
+    return response_data, can_cache
+
+
+def exec_dataset(dt: Dataset, xquery=None):
     """执行数据集
     返回 (sql语句, { rda})
     """
-    rdata = {}
+    full_sql = render_dataset_sql(dt, xquery)
-    results = {}
+    response_data, _ = execute_rendered_dataset(dt, full_sql)
-    results2 = {}
+    return full_sql, response_data
-    query = dt.default_param
-    if dt.sql_query:
-        query.update(xquery)
-        sql_f_ = check_sql_safe(dt.sql_query.format(**query))
-        sql_f_strip = sql_f_.strip(';')
-        sql_f_l = sql_f_strip.split(';')
-        # 多线程运行并返回字典结果
-        with concurrent.futures.ThreadPoolExecutor(max_workers=6) as executor:
-            fun_ps = []
-            for ind, val in enumerate(sql_f_l):
-                fun_ps.append((f'ds{ind}', execute_raw_sql, val))
-            # 生成执行函数
-            futures = {executor.submit(i[1], i[2]): i for i in fun_ps}
-            for future in concurrent.futures.as_completed(futures):
-                name, *_, sql_f = futures[future]  # 获取对应的键
-                res = future.result()
-                results[name],  results2[name] = format_sqldata(
-                    res[0], res[1])
-    rdata['data'] = results
-    rdata['data2'] = results2
-    return sql_f_, rdata

apps/bi/test_services.py

@@ -0,0 +1,105 @@
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from django.test import SimpleTestCase
+from rest_framework.exceptions import ParseError
+
+from apps.bi.services import (
+    exec_dataset,
+    execute_rendered_dataset,
+    render_dataset_sql,
+)
+from apps.bi.views import DatasetViewSet
+
+
+def dataset(**overrides):
+    values = {
+        "code": "output_daily",
+        "sql_query": "select * from output where day = '{day}'",
+        "default_param": {"day": "2026-08-01"},
+        "test_param": {"day": "2026-08-02"},
+        "echart_options": "",
+    }
+    values.update(overrides)
+    return SimpleNamespace(**values)
+
+
+class DatasetExecutionServiceTests(SimpleTestCase):
+    def test_render_does_not_mutate_default_parameters(self):
+        item = dataset()
+
+        sql = render_dataset_sql(item, {"day": "2026-08-10"})
+
+        self.assertIn("2026-08-10", sql)
+        self.assertEqual(item.default_param, {"day": "2026-08-01"})
+
+    def test_render_reports_missing_parameters(self):
+        item = dataset(default_param={}, sql_query="select '{required}'")
+
+        with self.assertRaises(ParseError):
+            render_dataset_sql(item)
+
+    def test_execute_formats_each_statement(self):
+        item = dataset(echart_options='{"series": {ds0}}')
+        with (
+            patch("apps.bi.services.execute_raw_sql", return_value=([], [])),
+            patch(
+                "apps.bi.services.format_sqldata",
+                return_value=([{"count": 1}], {"count": [1]}),
+            ),
+        ):
+            response, can_cache = execute_rendered_dataset(
+                item, "select 1;select 2"
+            )
+
+        self.assertTrue(can_cache)
+        self.assertEqual(set(response["data"]), {"ds0", "ds1"})
+        self.assertIn('"count": 1', response["echart_options"])
+
+    def test_empty_dataset_has_stable_empty_result(self):
+        full_sql, response = exec_dataset(dataset(sql_query=""))
+
+        self.assertEqual(full_sql, "")
+        self.assertEqual(response, {"data": {}, "data2": {}})
+
+
+class DatasetViewExecutionTests(SimpleTestCase):
+    @patch("apps.bi.views.cache")
+    @patch("apps.bi.views.execute_rendered_dataset")
+    @patch("apps.bi.views.render_dataset_sql")
+    @patch("apps.bi.views.DatasetSerializer")
+    def test_api_reuses_shared_execution_service(
+        self,
+        serializer_mock,
+        render_mock,
+        execute_mock,
+        cache_mock,
+    ):
+        item = dataset(
+            enabled=True,
+            name="日产量",
+            cache_seconds=10,
+        )
+        serializer_mock.return_value.data = {
+            "code": item.code,
+            "echart_options": "",
+        }
+        render_mock.return_value = "select 1"
+        execute_mock.return_value = (
+            {"data": {"ds0": [{"count": 1}]}, "data2": {}},
+            True,
+        )
+        cache_mock.get.return_value = None
+        view = DatasetViewSet(basename="dataset")
+        view.get_object = lambda: item
+        request = SimpleNamespace(
+            data={"query": {"day": "2026-08-10"}},
+            user=SimpleNamespace(id=42, belong_dept=SimpleNamespace(id=7)),
+        )
+
+        response = view.exec(request)
+
+        render_query = render_mock.call_args.args[1]
+        self.assertEqual(render_query["r_user"], 42)
+        self.assertEqual(render_query["r_dept"], 7)
+        self.assertEqual(response.data["data"]["ds0"][0]["count"], 1)

apps/bi/views.py

@@ -11,17 +11,13 @@ from apps.bi.serializers import (
     DatasetSerializer,
 )
 from django.apps import apps
-import concurrent.futures
 from django.core.cache import cache
-from apps.utils.sql import execute_raw_sql, format_sqldata
+from apps.bi.services import execute_rendered_dataset, render_dataset_sql
-from apps.bi.services import check_sql_safe, format_json_with_placeholders
 from rest_framework.exceptions import ParseError
 from rest_framework.generics import get_object_or_404
 from apps.utils.mixins import ListModelMixin
-import logging
 from drf_yasg import openapi
 from drf_yasg.utils import swagger_auto_schema
-myLogger = logging.getLogger('log')
 # Create your views here.
 
 
@@ -132,59 +128,24 @@ class DatasetViewSet(CustomModelViewSet):
         if not dt.enabled:
             raise ParseError(f'{dt.name}-该查询未启用')
         rdata = DatasetSerializer(instance=dt).data
-        xquery = request.data.get('query', {})
+        xquery = dict(request.data.get('query') or {})
         is_test = request.data.get('is_test', False)
         raise_exception = request.data.get('raise_exception', True)
         xquery['r_user'] = request.user.id
         xquery['r_dept'] = request.user.belong_dept.id if request.user.belong_dept else ''
-        can_cache = True
+        full_sql = render_dataset_sql(dt, xquery, is_test=is_test)
-        results = {}
+        hash_k = None
-        results2 = {}
+        if full_sql:
-        query = dt.default_param
+            sql_f_strip = full_sql.strip(';')
-        if dt.sql_query:
-            if is_test:
-                query.update(dt.test_param)
-            else:
-                query.update(xquery)
-            try:
-                sql_f_ = check_sql_safe(dt.sql_query.format(**query))
-            except KeyError as e:
-                raise ParseError(f'需指定查询参数_{str(e)}')
-            sql_f_strip = sql_f_.strip(';')
-            sql_f_l = sql_f_strip.split(';')
             hash_k = hash(sql_f_strip)
             hash_v = cache.get(hash_k, None)
             if hash_v:
                 return Response(hash_v)
-            # 多线程运行并返回字典结果
+        response_data, can_cache = execute_rendered_dataset(
-            with concurrent.futures.ThreadPoolExecutor(max_workers=6) as executor:
+            dt, full_sql, raise_exception=raise_exception
-                fun_ps = []
+        )
-                for ind, val in enumerate(sql_f_l):
+        rdata.update(response_data)
-                    fun_ps.append((f'ds{ind}', execute_raw_sql, val))
+        if response_data['data'] and can_cache and hash_k is not None:
-                # 生成执行函数
-                futures = {executor.submit(i[1], i[2]): i for i in fun_ps}
-                for future in concurrent.futures.as_completed(futures):
-                    name, *_, sql_f = futures[future]  # 获取对应的键
-                    try:
-                        res = future.result()
-                        results[name],  results2[name] = format_sqldata(
-                            res[0], res[1])
-                    except Exception as e:
-                        myLogger.error(f'bi查询异常：{str(e)}-{dt.code}--{sql_f}')
-                        if raise_exception:
-                            raise ParseError(f'查询异常：{str(e)}')
-                        else:
-                            results[name] = 'error: ' + str(e)
-                            can_cache = False
-        rdata['data'] = results
-        rdata['data2'] = results2
-        if rdata['echart_options'] and not rdata['echart_options'].startswith('function'):
-            for key in results:
-                if isinstance(results[key], str):
-                    raise ParseError(results[key])
-            rdata['echart_options'] = format_json_with_placeholders(
-                rdata['echart_options'], **results)
-        if results and can_cache:
             cache.set(hash_k, rdata, dt.cache_seconds)
         return Response(rdata)
 

docs/mcp.md

@@ -0,0 +1,49 @@
+# Factory MCP 服务
+
+Factory MCP 是仓库顶层的独立服务，使用官方 Python SDK v2，通过 Streamable HTTP 暴露 Agent 工具。当前已提供基础工具和第一批 Dataset 领域工具。
+
+## 启动
+
+在项目根目录使用项目虚拟环境启动独立进程：
+
+```powershell
+.venv\Scripts\python.exe -m mcp_server
+```
+
+默认监听 `127.0.0.1:2260`，MCP 端点为 `/mcp`。客户端必须在每次请求中携带 Factory access token：
+
+```text
+Authorization: Bearer <Factory access token>
+```
+
+## 配置
+
+生产环境在本机已忽略的 `config/conf.py` 中覆盖以下配置：
+
+- `MCP_HOST`：监听地址。
+- `MCP_PORT`：监听端口。
+- `MCP_PATH`：Streamable HTTP 路径。
+- `MCP_ALLOWED_HOSTS`：允许的 HTTP Host，支持 `hostname:*` 端口通配形式。
+- `MCP_ALLOWED_ORIGINS`：允许的浏览器 Origin；非浏览器客户端通常不发送 Origin。
+- `MCP_MAX_REQUEST_BODY_SIZE`：单个 MCP 请求体上限，默认 1 MiB。
+- `MCP_MAX_RESULT_BYTES`：单次领域工具结果上限，默认 512 KiB。
+
+生产部署必须明确配置实际域名的 Host 白名单，不应直接复用 Django 当前的宽泛 `ALLOWED_HOSTS`。
+
+## 基础工具
+
+- `factory_server_info`：返回系统版本、MCP 协议版本和认证方式。
+- `factory_whoami`：返回当前 JWT 对应的 Factory 用户。
+- `search_datasets`：按名称、code 或描述搜索启用的数据集，不返回 SQL 配置。
+- `execute_dataset`：按 code 执行数据集，需要当前用户具有 `dataset.exec` 权限。
+- `search_wprs`：按编号、物料、批次、状态和当前位置搜索 WPR，只返回摘要。
+- `get_wpr`：按 ID、内部编号或对外编号读取 WPR 详情、缺陷和业务数据。
+
+WPR 当前沿用既有 API 的读取边界：有效登录用户可读，且该 ViewSet 未启用部门数据过滤。MCP 不开放修改编号、分配对外编号或更新预处理信息等写操作。
+
+- `search_batch_stats`：按批次、直通大批、起始物料和版本搜索批次统计摘要。
+- `get_batch_stat`：读取指定批次版本的完整统计数据，可附带直接拆批/合批关系。
+
+BatchSt 同样沿用既有 API 的 `get: *` 读取边界，不提供创建、重算或修改工具。完整统计结果仍受 `MCP_MAX_RESULT_BYTES` 限制。
+
+新增领域工具时必须从 MCP 请求身份获取用户，并复用 Factory 的权限码和数据范围过滤；不得直接使用固定管理员身份查询 ORM。

mcp_server/__init__.py

@@ -0,0 +1 @@
+"""Factory MCP v2 integration."""

mcp_server/__main__.py

@@ -0,0 +1,24 @@
+import os
+
+import django
+import uvicorn
+
+
+def main() -> None:
+    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "server.settings")
+    django.setup()
+
+    from mcp_server.server import application
+
+    from django.conf import settings
+
+    uvicorn.run(
+        application,
+        host=settings.MCP_HOST,
+        port=settings.MCP_PORT,
+        log_level="info",
+    )
+
+
+if __name__ == "__main__":
+    main()

mcp_server/auth.py

@@ -0,0 +1,66 @@
+from asgiref.sync import sync_to_async
+from rest_framework.exceptions import APIException
+from rest_framework_simplejwt.authentication import JWTAuthentication
+from rest_framework_simplejwt.exceptions import TokenError
+
+from mcp.server.auth.provider import AccessToken
+from mcp.server.auth.middleware.bearer_auth import RequireAuthMiddleware
+from starlette.types import ASGIApp, Receive, Scope, Send
+
+
+def verify_factory_jwt(token: str) -> AccessToken | None:
+    """验证 Factory access token，并生成 MCP 的逐请求身份信息。"""
+    authentication = JWTAuthentication()
+    try:
+        validated_token = authentication.get_validated_token(token)
+        user = authentication.get_user(validated_token)
+    except (APIException, TokenError):
+        return None
+
+    return AccessToken(
+        token=token,
+        client_id="factory-mcp",
+        scopes=["factory:user"],
+        expires_at=validated_token.get("exp"),
+        subject=str(user.pk),
+        claims={
+            "factory_user": {
+                "id": str(user.pk),
+                "username": user.get_username(),
+                "name": user.name,
+                "is_superuser": user.is_superuser,
+            },
+        },
+    )
+
+
+class FactoryJWTVerifier:
+    """让 MCP SDK 复用 Factory SimpleJWT 的验证规则。"""
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        return await sync_to_async(
+            verify_factory_jwt,
+            thread_sensitive=True,
+        )(token)
+
+
+class RequireFactoryJWTMiddleware:
+    """仅保护 HTTP 请求，并把 ASGI lifespan 原样交给 MCP SDK。"""
+
+    def __init__(self, app: ASGIApp):
+        self.app = app
+        self.protected_app = RequireAuthMiddleware(
+            app,
+            required_scopes=[],
+        )
+
+    async def __call__(
+        self,
+        scope: Scope,
+        receive: Receive,
+        send: Send,
+    ) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+        await self.protected_app(scope, receive, send)

mcp_server/context.py

@@ -0,0 +1,32 @@
+from typing import Any
+
+from django.contrib.auth import get_user_model
+from mcp.server.auth.middleware.auth_context import get_access_token
+
+from apps.utils.permission import has_perm
+
+
+def authenticated_user_claims() -> dict[str, Any]:
+    """返回当前请求中经过 Factory JWT 校验的用户摘要。"""
+    access_token = get_access_token()
+    claims = access_token.claims if access_token else None
+    user = claims.get("factory_user") if claims else None
+    if not isinstance(user, dict):
+        raise RuntimeError("当前 MCP 请求缺少有效的 Factory 用户身份")
+    return user
+
+
+def authenticated_factory_user():
+    """加载当前 JWT 对应的 Django 用户，供权限和数据范围逻辑复用。"""
+    claims = authenticated_user_claims()
+    try:
+        return get_user_model().objects.select_related("belong_dept").get(
+            pk=claims["id"]
+        )
+    except (KeyError, get_user_model().DoesNotExist) as exc:
+        raise RuntimeError("当前 JWT 对应的 Factory 用户不存在") from exc
+
+
+def require_permission(user, permission_code: str) -> None:
+    if not has_perm(user, [permission_code]):
+        raise PermissionError(f"当前用户缺少权限：{permission_code}")

mcp_server/server.py

@@ -0,0 +1,89 @@
+from collections.abc import Sequence
+from typing import Any
+
+from django.conf import settings
+from mcp.server import MCPServer
+from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
+from mcp.server.auth.middleware.bearer_auth import (
+    BearerAuthBackend,
+)
+from mcp.server.auth.provider import TokenVerifier
+from mcp.server.transport_security import TransportSecuritySettings
+from starlette.middleware.authentication import AuthenticationMiddleware
+from starlette.types import ASGIApp
+
+from mcp_server.auth import (
+    FactoryJWTVerifier,
+    RequireFactoryJWTMiddleware,
+)
+from mcp_server.context import authenticated_user_claims
+from mcp_server.tools.batch_stats import register_batch_stat_tools
+from mcp_server.tools.datasets import register_dataset_tools
+from mcp_server.tools.wprs import register_wpr_tools
+
+
+PROTOCOL_REVISION = "2026-07-28"
+
+mcp = MCPServer(
+    name="factory",
+    title="Factory MCP",
+    description="Factory 面向 Agent 的受控业务能力入口。",
+    instructions="所有工具均使用当前请求携带的 Factory JWT 身份执行。",
+    version=settings.SYS_VERSION,
+)
+
+
+@mcp.tool()
+def factory_server_info() -> dict[str, Any]:
+    """返回 Factory MCP 服务版本及协议基础信息。"""
+    return {
+        "name": "factory",
+        "system_version": settings.SYS_VERSION,
+        "protocol_revision": PROTOCOL_REVISION,
+        "authentication": "factory_jwt",
+        "domain_tools_ready": True,
+    }
+
+
+@mcp.tool()
+def factory_whoami() -> dict[str, Any]:
+    """返回当前 Factory JWT 对应的用户身份。"""
+    return authenticated_user_claims()
+
+
+register_dataset_tools(mcp)
+register_wpr_tools(mcp)
+register_batch_stat_tools(mcp)
+
+
+def create_app(
+    *,
+    allowed_hosts: Sequence[str] | None = None,
+    allowed_origins: Sequence[str] | None = None,
+    token_verifier: TokenVerifier | None = None,
+) -> ASGIApp:
+    """创建仅接受 Factory JWT 的 MCP v2 Streamable HTTP 应用。"""
+    transport_security = TransportSecuritySettings(
+        enable_dns_rebinding_protection=True,
+        allowed_hosts=list(settings.MCP_ALLOWED_HOSTS if allowed_hosts is None else allowed_hosts),
+        allowed_origins=list(settings.MCP_ALLOWED_ORIGINS if allowed_origins is None else allowed_origins),
+    )
+    app: ASGIApp = mcp.streamable_http_app(
+        streamable_http_path=settings.MCP_PATH,
+        json_response=True,
+        max_request_body_size=settings.MCP_MAX_REQUEST_BODY_SIZE,
+        transport_security=transport_security,
+        host=settings.MCP_HOST,
+    )
+
+    # 包装顺序保证先解析 Bearer JWT，再写入 MCP 请求上下文，最后强制认证。
+    app = RequireFactoryJWTMiddleware(app)
+    app = AuthContextMiddleware(app)
+    app = AuthenticationMiddleware(
+        app,
+        backend=BearerAuthBackend(token_verifier or FactoryJWTVerifier()),
+    )
+    return app
+
+
+application = create_app()

mcp_server/test_batch_stats.py

@@ -0,0 +1,105 @@
+from datetime import datetime
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from django.test import SimpleTestCase
+
+from apps.wpm.models import BatchSt
+from mcp_server.tools.batch_stats import get_batch_stat, search_batch_stats
+
+
+def batch_stat(**overrides):
+    values = {
+        "id": "500",
+        "batch": "BATCH-001",
+        "version": 1,
+        "zt_batch": "ZT-001",
+        "first_time": datetime(2026, 8, 1, 8, 0),
+        "last_time": datetime(2026, 8, 2, 8, 0),
+        "material_start": SimpleNamespace(
+            id="100",
+            name="原料",
+            model="M-1",
+            specification="S-1",
+        ),
+        "data": {"output": {"count": 10}, "quality": {"ok": 9}},
+        "update_time": datetime(2026, 8, 10, 8, 0),
+    }
+    values.update(overrides)
+    return SimpleNamespace(**values)
+
+
+class BatchStatToolTests(SimpleTestCase):
+    @patch("mcp_server.tools.batch_stats._base_queryset")
+    @patch("mcp_server.tools.batch_stats.authenticated_factory_user")
+    def test_search_returns_summary_without_full_data(
+        self,
+        _user_mock,
+        queryset_mock,
+    ):
+        queryset = MagicMock()
+        queryset.filter.return_value = queryset
+        queryset.order_by.return_value = queryset
+        queryset.__getitem__.return_value = [batch_stat()]
+        queryset_mock.return_value = queryset
+
+        result = search_batch_stats(query="BATCH", limit=10)
+
+        self.assertEqual(result["items"][0]["batch"], "BATCH-001")
+        self.assertEqual(result["items"][0]["data_keys"], ["output", "quality"])
+        self.assertNotIn("data", result["items"][0])
+
+    @patch("mcp_server.tools.batch_stats.BatchLog.objects.filter")
+    @patch("mcp_server.tools.batch_stats._base_queryset")
+    @patch("mcp_server.tools.batch_stats.authenticated_factory_user")
+    def test_get_returns_data_and_direct_relations(
+        self,
+        _user_mock,
+        queryset_mock,
+        relation_filter_mock,
+    ):
+        item = batch_stat()
+        queryset_mock.return_value.get.return_value = item
+        relation_filter_mock.return_value.select_related.return_value.values.return_value = [
+            {
+                "id": "600",
+                "relation_type": "split",
+                "source_id": "500",
+                "source__batch": "BATCH-001",
+                "source__version": 1,
+                "target_id": "501",
+                "target__batch": "BATCH-001-1",
+                "target__version": 1,
+                "handover_id": "700",
+                "mlog_id": None,
+            }
+        ]
+
+        result = get_batch_stat("BATCH-001")
+
+        self.assertEqual(result["data"]["output"]["count"], 10)
+        self.assertEqual(result["relations"][0]["relation_type"], "split")
+
+    @patch("mcp_server.tools.batch_stats.BatchLog.objects.filter")
+    @patch("mcp_server.tools.batch_stats._base_queryset")
+    @patch("mcp_server.tools.batch_stats.authenticated_factory_user")
+    def test_get_can_omit_relations(
+        self,
+        _user_mock,
+        queryset_mock,
+        relation_filter_mock,
+    ):
+        queryset_mock.return_value.get.return_value = batch_stat()
+
+        result = get_batch_stat("BATCH-001", include_relations=False)
+
+        self.assertNotIn("relations", result)
+        relation_filter_mock.assert_not_called()
+
+    @patch("mcp_server.tools.batch_stats._base_queryset")
+    @patch("mcp_server.tools.batch_stats.authenticated_factory_user")
+    def test_get_reports_missing_batch(self, _user_mock, queryset_mock):
+        queryset_mock.return_value.get.side_effect = BatchSt.DoesNotExist
+
+        with self.assertRaisesRegex(ValueError, "未找到批次统计"):
+            get_batch_stat("missing")

mcp_server/test_datasets.py

@@ -0,0 +1,96 @@
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from django.test import SimpleTestCase, override_settings
+
+from mcp_server.tools.datasets import execute_dataset, search_datasets
+
+
+class DatasetToolTests(SimpleTestCase):
+    @patch("mcp_server.tools.datasets.authenticated_factory_user")
+    @patch("mcp_server.tools.datasets.Dataset.objects.filter")
+    def test_search_returns_safe_catalog_fields(self, filter_mock, _user_mock):
+        queryset = MagicMock()
+        filter_mock.return_value = queryset
+        queryset.order_by.return_value.values.return_value.__getitem__.return_value = [
+            {
+                "code": "daily_output",
+                "name": "日产量",
+                "description": "按日统计产量",
+                "default_param": {"day": "2026-08-10"},
+                "test_param": {},
+            }
+        ]
+
+        result = search_datasets(limit=10)
+
+        self.assertEqual(result["items"][0]["code"], "daily_output")
+        self.assertNotIn("sql_query", result["items"][0])
+
+    @patch("mcp_server.tools.datasets.cache")
+    @patch("mcp_server.tools.datasets.execute_rendered_dataset")
+    @patch("mcp_server.tools.datasets.render_dataset_sql")
+    @patch("mcp_server.tools.datasets.require_permission")
+    @patch("mcp_server.tools.datasets.authenticated_factory_user")
+    @patch("mcp_server.tools.datasets.Dataset.objects.get")
+    def test_execute_reuses_identity_permission_and_service(
+        self,
+        get_mock,
+        user_mock,
+        permission_mock,
+        render_mock,
+        execute_mock,
+        cache_mock,
+    ):
+        item = SimpleNamespace(
+            code="daily_output",
+            name="日产量",
+            description="按日统计产量",
+            cache_seconds=10,
+        )
+        user = SimpleNamespace(id=42, belong_dept_id=7)
+        get_mock.return_value = item
+        user_mock.return_value = user
+        render_mock.return_value = "select 1"
+        cache_mock.get.return_value = None
+        execute_mock.return_value = (
+            {"data": {"ds0": [{"count": 1}]}, "data2": {}},
+            True,
+        )
+
+        result = execute_dataset("daily_output", {"day": "2026-08-10"})
+
+        permission_mock.assert_called_once_with(user, "dataset.exec")
+        render_query = render_mock.call_args.args[1]
+        self.assertEqual(render_query["r_user"], 42)
+        self.assertEqual(render_query["r_dept"], 7)
+        self.assertEqual(result["data"]["ds0"][0]["count"], 1)
+        self.assertNotIn("sql_query", result)
+
+    @override_settings(MCP_MAX_RESULT_BYTES=1)
+    @patch("mcp_server.tools.datasets.cache")
+    @patch("mcp_server.tools.datasets.execute_rendered_dataset")
+    @patch("mcp_server.tools.datasets.render_dataset_sql", return_value="")
+    @patch("mcp_server.tools.datasets.require_permission")
+    @patch("mcp_server.tools.datasets.authenticated_factory_user")
+    @patch("mcp_server.tools.datasets.Dataset.objects.get")
+    def test_execute_rejects_oversized_results(
+        self,
+        get_mock,
+        user_mock,
+        _permission_mock,
+        _render_mock,
+        execute_mock,
+        _cache_mock,
+    ):
+        get_mock.return_value = SimpleNamespace(
+            code="daily_output",
+            name="日产量",
+            description="",
+            cache_seconds=0,
+        )
+        user_mock.return_value = SimpleNamespace(id=42, belong_dept_id=None)
+        execute_mock.return_value = ({"data": {"ds0": [1]}, "data2": {}}, True)
+
+        with self.assertRaisesRegex(RuntimeError, "超过 MCP 响应上限"):
+            execute_dataset("daily_output")

mcp_server/test_wprs.py

@@ -0,0 +1,103 @@
+from datetime import datetime
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from django.test import SimpleTestCase
+
+from mcp_server.tools.wprs import get_wpr, search_wprs
+
+
+def material(material_id="100", name="成品"):
+    return SimpleNamespace(
+        id=material_id,
+        name=name,
+        model="M-1",
+        specification="S-1",
+    )
+
+
+def wpr(**overrides):
+    values = {
+        "id": "200",
+        "number": "WPR-001",
+        "number_out": "OUT-001",
+        "version": 1,
+        "state": 10,
+        "get_state_display": lambda: "正常",
+        "material": material(),
+        "material_start": material("101", "原料"),
+        "wm_id": "300",
+        "wm": SimpleNamespace(batch="WP-001"),
+        "mb_id": None,
+        "mb": None,
+        "wpr_from_id": None,
+        "wpr_from": None,
+        "oinfo": {"test": "ok"},
+        "data": {"route": []},
+        "pre_info": {"tooling": "T-1"},
+        "create_time": datetime(2026, 8, 1, 8, 0),
+        "update_time": datetime(2026, 8, 10, 8, 0),
+    }
+    values.update(overrides)
+    return SimpleNamespace(**values)
+
+
+class WprToolTests(SimpleTestCase):
+    @patch("mcp_server.tools.wprs._base_queryset")
+    @patch("mcp_server.tools.wprs.authenticated_factory_user")
+    def test_search_returns_read_only_summary(self, _user_mock, queryset_mock):
+        queryset = MagicMock()
+        queryset.filter.return_value = queryset
+        queryset.distinct.return_value = queryset
+        queryset.order_by.return_value = queryset
+        queryset.__getitem__.return_value = [wpr()]
+        queryset_mock.return_value = queryset
+
+        result = search_wprs(query="WPR", location="workshop", limit=10)
+
+        self.assertEqual(result["items"][0]["number"], "WPR-001")
+        self.assertEqual(result["items"][0]["workshop_batch"], "WP-001")
+        self.assertNotIn("data", result["items"][0])
+        self.assertNotIn("pre_info", result["items"][0])
+
+    @patch("mcp_server.tools.wprs.WprDefect.objects.filter")
+    @patch("mcp_server.tools.wprs._base_queryset")
+    @patch("mcp_server.tools.wprs.authenticated_factory_user")
+    def test_get_returns_business_detail(
+        self,
+        _user_mock,
+        queryset_mock,
+        defect_filter_mock,
+    ):
+        item = wpr()
+        queryset = MagicMock()
+        queryset.filter.return_value.order_by.return_value.first.return_value = item
+        queryset_mock.return_value = queryset
+        defect_filter_mock.return_value.select_related.return_value.values.return_value = [
+            {
+                "defect_id": "400",
+                "defect__name": "划伤",
+                "is_main": True,
+            }
+        ]
+
+        result = get_wpr("WPR-001")
+
+        self.assertEqual(result["material"]["name"], "成品")
+        self.assertEqual(result["defects"][0]["defect__name"], "划伤")
+        self.assertEqual(result["pre_info"]["tooling"], "T-1")
+
+    @patch("mcp_server.tools.wprs._base_queryset")
+    @patch("mcp_server.tools.wprs.authenticated_factory_user")
+    def test_get_reports_missing_wpr(self, _user_mock, queryset_mock):
+        queryset = MagicMock()
+        queryset.filter.return_value.order_by.return_value.first.return_value = None
+        queryset_mock.return_value = queryset
+
+        with self.assertRaisesRegex(ValueError, "未找到 WPR"):
+            get_wpr("missing")
+
+    @patch("mcp_server.tools.wprs.authenticated_factory_user")
+    def test_search_rejects_unknown_location(self, _user_mock):
+        with self.assertRaisesRegex(ValueError, "不支持的 WPR 位置"):
+            search_wprs(location="invalid")

mcp_server/tests.py

@@ -0,0 +1,164 @@
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from django.test import SimpleTestCase
+from mcp.server.auth.middleware.auth_context import auth_context_var
+from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
+from mcp.server.auth.provider import AccessToken
+from rest_framework.exceptions import AuthenticationFailed
+from starlette.testclient import TestClient
+
+from mcp_server.auth import verify_factory_jwt
+from mcp_server.server import (
+    PROTOCOL_REVISION,
+    create_app,
+    factory_server_info,
+    factory_whoami,
+    mcp,
+)
+
+
+class FactoryJWTVerifierTests(SimpleTestCase):
+    def test_factory_access_token_is_accepted(self):
+        user = SimpleNamespace(
+            pk=42,
+            name="MCP用户",
+            is_superuser=False,
+            get_username=lambda: "mcp-user",
+        )
+        authentication = patch("mcp_server.auth.JWTAuthentication").start()
+        self.addCleanup(patch.stopall)
+        authentication.return_value.get_validated_token.return_value = {
+            "exp": 1234567890,
+        }
+        authentication.return_value.get_user.return_value = user
+
+        access_token = verify_factory_jwt("access-token")
+
+        self.assertIsNotNone(access_token)
+        self.assertEqual(access_token.subject, str(user.pk))
+        self.assertEqual(
+            access_token.claims["factory_user"]["username"],
+            user.get_username(),
+        )
+
+    def test_invalid_token_is_rejected(self):
+        with patch("mcp_server.auth.JWTAuthentication") as authentication:
+            authentication.return_value.get_validated_token.side_effect = AuthenticationFailed("invalid token")
+
+            self.assertIsNone(verify_factory_jwt("invalid-token"))
+
+
+class FactoryMCPServerTests(SimpleTestCase):
+    def test_base_tools_are_registered(self):
+        tools = asyncio.run(mcp.list_tools())
+        names = {tool.name for tool in tools}
+
+        self.assertEqual(
+            names,
+            {
+                "execute_dataset",
+                "factory_server_info",
+                "factory_whoami",
+                "get_batch_stat",
+                "get_wpr",
+                "search_batch_stats",
+                "search_datasets",
+                "search_wprs",
+            },
+        )
+
+    def test_server_info_targets_mcp_v2(self):
+        result = factory_server_info()
+
+        self.assertEqual(result["protocol_revision"], PROTOCOL_REVISION)
+        self.assertTrue(result["domain_tools_ready"])
+
+    def test_whoami_uses_authenticated_request_context(self):
+        user = {
+            "id": "42",
+            "username": "agent-user",
+            "name": "Agent用户",
+            "is_superuser": False,
+        }
+        authenticated = AuthenticatedUser(
+            AccessToken(
+                token="test-token",
+                client_id="factory-mcp",
+                scopes=["factory:user"],
+                subject=user["id"],
+                claims={"factory_user": user},
+            )
+        )
+        context_token = auth_context_var.set(authenticated)
+        try:
+            self.assertEqual(factory_whoami(), user)
+        finally:
+            auth_context_var.reset(context_token)
+
+    def test_http_endpoint_requires_bearer_token(self):
+        app = create_app(allowed_hosts=["testserver"])
+
+        with TestClient(app) as client:
+            response = client.post("/mcp", json={})
+
+        self.assertEqual(response.status_code, 401)
+        self.assertEqual(response.json()["error"], "invalid_token")
+
+    def test_mcp_v2_request_uses_bearer_identity(self):
+        user = {
+            "id": "42",
+            "username": "agent-user",
+            "name": "Agent用户",
+            "is_superuser": False,
+        }
+
+        class TestTokenVerifier:
+            async def verify_token(self, token):
+                if token != "valid-token":
+                    return None
+                return AccessToken(
+                    token=token,
+                    client_id="test-client",
+                    scopes=["factory:user"],
+                    subject=user["id"],
+                    claims={"factory_user": user},
+                )
+
+        app = create_app(
+            allowed_hosts=["testserver"],
+            token_verifier=TestTokenVerifier(),
+        )
+        request = {
+            "jsonrpc": "2.0",
+            "id": 1,
+            "method": "tools/call",
+            "params": {
+                "name": "factory_whoami",
+                "arguments": {},
+                "_meta": {
+                    "io.modelcontextprotocol/protocolVersion": (PROTOCOL_REVISION),
+                    "io.modelcontextprotocol/clientInfo": {
+                        "name": "factory-tests",
+                        "version": "1.0",
+                    },
+                    "io.modelcontextprotocol/clientCapabilities": {},
+                },
+            },
+        }
+        headers = {
+            "Authorization": "Bearer valid-token",
+            "MCP-Protocol-Version": PROTOCOL_REVISION,
+            "Mcp-Method": "tools/call",
+            "Mcp-Name": "factory_whoami",
+        }
+
+        with TestClient(app) as client:
+            response = client.post("/mcp", json=request, headers=headers)
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(
+            response.json()["result"]["structuredContent"],
+            user,
+        )

mcp_server/tools/__init__.py

@@ -0,0 +1 @@
+"""Factory MCP 领域工具，按业务域拆分并在 server 中显式注册。"""

mcp_server/tools/batch_stats.py

@@ -0,0 +1,104 @@
+from typing import Any
+
+from django.db.models import Q
+
+from apps.wpm.models import BatchLog, BatchSt
+from mcp_server.context import authenticated_factory_user
+from mcp_server.tools.common import json_safe_result, validate_result_size
+
+
+def _base_queryset():
+    return BatchSt.objects.select_related("material_start")
+
+
+def _batch_summary(batch_stat: BatchSt) -> dict[str, Any]:
+    material = batch_stat.material_start
+    return {
+        "id": str(batch_stat.id),
+        "batch": batch_stat.batch,
+        "version": batch_stat.version,
+        "zt_batch": batch_stat.zt_batch,
+        "first_time": batch_stat.first_time,
+        "last_time": batch_stat.last_time,
+        "material_start": (
+            {
+                "id": str(material.id),
+                "name": material.name,
+                "model": material.model,
+                "specification": material.specification,
+            }
+            if material
+            else None
+        ),
+        "data_keys": sorted((batch_stat.data or {}).keys()),
+        "update_time": batch_stat.update_time,
+    }
+
+
+def search_batch_stats(
+    query: str = "",
+    zt_batch: str = "",
+    material_id: str | None = None,
+    version: int | None = 1,
+    limit: int = 20,
+) -> dict[str, Any]:
+    """搜索批次统计；摘要仅返回数据分组名称，不返回完整统计数据。"""
+    authenticated_factory_user()
+    safe_limit = max(1, min(limit, 100))
+    queryset = _base_queryset()
+    if query.strip():
+        queryset = queryset.filter(batch__icontains=query.strip())
+    if zt_batch.strip():
+        queryset = queryset.filter(zt_batch=zt_batch.strip())
+    if material_id:
+        queryset = queryset.filter(material_start_id=material_id)
+    if version is not None:
+        queryset = queryset.filter(version=version)
+    items = [
+        _batch_summary(item)
+        for item in queryset.order_by("batch", "version")[:safe_limit]
+    ]
+    result = json_safe_result({"items": items, "limit": safe_limit})
+    validate_result_size(result)
+    return result
+
+
+def get_batch_stat(
+    batch: str,
+    version: int = 1,
+    include_relations: bool = True,
+) -> dict[str, Any]:
+    """读取指定批次版本的完整统计数据，并可附带直接拆合批关系。"""
+    authenticated_factory_user()
+    try:
+        batch_stat = _base_queryset().get(batch=batch, version=version)
+    except BatchSt.DoesNotExist as exc:
+        raise ValueError(f"未找到批次统计：{batch} v{version}") from exc
+
+    result = _batch_summary(batch_stat)
+    result["data"] = batch_stat.data
+    if include_relations:
+        result["relations"] = list(
+            BatchLog.objects.filter(Q(source=batch_stat) | Q(target=batch_stat))
+            .select_related("source", "target")
+            .values(
+                "id",
+                "relation_type",
+                "source_id",
+                "source__batch",
+                "source__version",
+                "target_id",
+                "target__batch",
+                "target__version",
+                "handover_id",
+                "mlog_id",
+            )
+        )
+    result = json_safe_result(result)
+    validate_result_size(result)
+    return result
+
+
+def register_batch_stat_tools(server) -> None:
+    server.tool()(search_batch_stats)
+    server.tool()(get_batch_stat)

mcp_server/tools/common.py

@@ -0,0 +1,18 @@
+import json
+from typing import Any
+
+from django.conf import settings
+
+from apps.utils.tools import MyJSONEncoder
+
+
+def json_safe_result(value: Any) -> Any:
+    return json.loads(json.dumps(value, cls=MyJSONEncoder, ensure_ascii=False))
+
+
+def validate_result_size(value: Any) -> None:
+    encoded = json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode()
+    if len(encoded) > settings.MCP_MAX_RESULT_BYTES:
+        raise RuntimeError(
+            "工具结果超过 MCP 响应上限，请缩小查询范围或增加筛选参数"
+        )

mcp_server/tools/datasets.py

@@ -0,0 +1,79 @@
+import hashlib
+from typing import Any
+
+from django.core.cache import cache
+from django.db.models import Q
+
+from apps.bi.models import Dataset
+from apps.bi.services import execute_rendered_dataset, render_dataset_sql
+from mcp_server.context import authenticated_factory_user, require_permission
+from mcp_server.tools.common import json_safe_result, validate_result_size
+
+
+def search_datasets(query: str = "", limit: int = 20) -> dict[str, Any]:
+    """搜索可执行的数据集目录，不返回 SQL 等敏感配置。"""
+    authenticated_factory_user()
+    safe_limit = max(1, min(limit, 100))
+    queryset = Dataset.objects.filter(enabled=True)
+    if query.strip():
+        queryset = queryset.filter(
+            Q(name__icontains=query.strip())
+            | Q(code__icontains=query.strip())
+            | Q(description__icontains=query.strip())
+        )
+    rows = queryset.order_by("name", "code", "id").values(
+        "code",
+        "name",
+        "description",
+        "default_param",
+        "test_param",
+    )[:safe_limit]
+    return {"items": list(rows), "limit": safe_limit}
+
+
+def execute_dataset(
+    code: str,
+    parameters: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """以当前 Factory 用户身份执行启用的数据集。需要 dataset.exec 权限。"""
+    user = authenticated_factory_user()
+    require_permission(user, "dataset.exec")
+    try:
+        dataset = Dataset.objects.get(code=code, enabled=True)
+    except Dataset.DoesNotExist as exc:
+        raise ValueError(f"未找到已启用的数据集：{code}") from exc
+    except Dataset.MultipleObjectsReturned as exc:
+        raise RuntimeError(f"数据集 code 不唯一，无法执行：{code}") from exc
+
+    query = dict(parameters or {})
+    query["r_user"] = user.id
+    query["r_dept"] = user.belong_dept_id or ""
+    full_sql = render_dataset_sql(dataset, query)
+
+    cache_key = None
+    response_data = None
+    if full_sql and dataset.cache_seconds:
+        digest = hashlib.sha256(full_sql.strip(";").encode()).hexdigest()
+        cache_key = f"mcp:dataset:{digest}"
+        response_data = cache.get(cache_key)
+
+    if response_data is None:
+        response_data, can_cache = execute_rendered_dataset(dataset, full_sql)
+        if cache_key and can_cache and response_data["data"]:
+            cache.set(cache_key, response_data, dataset.cache_seconds)
+
+    result = json_safe_result(
+        {
+            "code": dataset.code,
+            "name": dataset.name,
+            "description": dataset.description,
+            **response_data,
+        }
+    )
+    validate_result_size(result)
+    return result
+
+
+def register_dataset_tools(server) -> None:
+    server.tool()(search_datasets)
+    server.tool()(execute_dataset)

mcp_server/tools/wprs.py

@@ -0,0 +1,139 @@
+from typing import Any, Literal
+
+from django.db.models import Q
+
+from apps.wpmw.models import Wpr, WprDefect
+from mcp_server.context import authenticated_factory_user
+from mcp_server.tools.common import json_safe_result, validate_result_size
+
+
+WprLocation = Literal["all", "workshop", "warehouse", "unassigned"]
+
+
+def _base_queryset():
+    return Wpr.objects.select_related(
+        "material",
+        "material_start",
+        "wm",
+        "mb",
+        "wpr_from",
+    )
+
+
+def _wpr_summary(wpr: Wpr) -> dict[str, Any]:
+    material = wpr.material
+    return {
+        "id": str(wpr.id),
+        "number": wpr.number,
+        "number_out": wpr.number_out,
+        "version": wpr.version,
+        "state": wpr.state,
+        "state_name": wpr.get_state_display(),
+        "material": {
+            "id": str(material.id),
+            "name": material.name,
+            "model": material.model,
+            "specification": material.specification,
+        },
+        "workshop_batch": wpr.wm.batch if wpr.wm_id else None,
+        "warehouse_batch": wpr.mb.batch if wpr.mb_id else None,
+        "create_time": wpr.create_time,
+        "update_time": wpr.update_time,
+    }
+
+
+def search_wprs(
+    query: str = "",
+    state: int | None = None,
+    material_id: str | None = None,
+    batch: str = "",
+    location: WprLocation = "all",
+    limit: int = 20,
+) -> dict[str, Any]:
+    """按编号、物料或批次搜索单件产品；仅提供只读摘要。"""
+    authenticated_factory_user()
+    if location not in {"all", "workshop", "warehouse", "unassigned"}:
+        raise ValueError(f"不支持的 WPR 位置：{location}")
+    safe_limit = max(1, min(limit, 100))
+    queryset = _base_queryset()
+    if query.strip():
+        keyword = query.strip()
+        queryset = queryset.filter(
+            Q(number__icontains=keyword)
+            | Q(number_out__icontains=keyword)
+            | Q(material__name__icontains=keyword)
+            | Q(material__model__icontains=keyword)
+            | Q(material__specification__icontains=keyword)
+        )
+    if state is not None:
+        queryset = queryset.filter(state=state)
+    if material_id:
+        queryset = queryset.filter(material_id=material_id)
+    if batch.strip():
+        queryset = queryset.filter(
+            Q(wm__batch__icontains=batch.strip())
+            | Q(mb__batch__icontains=batch.strip())
+        )
+    if location == "workshop":
+        queryset = queryset.filter(wm__isnull=False)
+    elif location == "warehouse":
+        queryset = queryset.filter(mb__isnull=False)
+    elif location == "unassigned":
+        queryset = queryset.filter(wm__isnull=True, mb__isnull=True)
+
+    items = [
+        _wpr_summary(wpr)
+        for wpr in queryset.distinct().order_by("number", "create_time")[:safe_limit]
+    ]
+    result = json_safe_result({"items": items, "limit": safe_limit})
+    validate_result_size(result)
+    return result
+
+
+def get_wpr(identifier: str) -> dict[str, Any]:
+    """按 WPR ID、内部编号或对外编号读取单件详情。"""
+    authenticated_factory_user()
+    lookup = Q(number=identifier) | Q(number_out=identifier)
+    if identifier.isdigit():
+        lookup |= Q(pk=identifier)
+    wpr = _base_queryset().filter(lookup).order_by("-version", "-update_time").first()
+    if wpr is None:
+        raise ValueError(f"未找到 WPR：{identifier}")
+
+    result = _wpr_summary(wpr)
+    material_start = wpr.material_start
+    result.update(
+        {
+            "material_start": (
+                {
+                    "id": str(material_start.id),
+                    "name": material_start.name,
+                    "model": material_start.model,
+                    "specification": material_start.specification,
+                }
+                if material_start
+                else None
+            ),
+            "wpr_from": (
+                {"id": str(wpr.wpr_from.id), "number": wpr.wpr_from.number}
+                if wpr.wpr_from_id
+                else None
+            ),
+            "oinfo": wpr.oinfo,
+            "data": wpr.data,
+            "pre_info": wpr.pre_info,
+            "defects": list(
+                WprDefect.objects.filter(wpr=wpr)
+                .select_related("defect")
+                .values("defect_id", "defect__name", "is_main")
+            ),
+        }
+    )
+    result = json_safe_result(result)
+    validate_result_size(result)
+    return result
+
+
+def register_wpr_tools(server) -> None:
+    server.tool()(search_wprs)
+    server.tool()(get_wpr)

requirements.txt

@@ -10,6 +10,11 @@ django-cors-headers==4.9.0
 djangorestframework-simplejwt==5.5.1
 django-restql==0.15.2
 
+# =======================
+# Agent Integration
+# =======================
+mcp==2.0.0
+
 # =======================
 # Celery
 # =======================

server/settings.py

@@ -241,6 +241,20 @@ SIMPLE_JWT = {
     'REFRESH_TOKEN_LIFETIME': timedelta(days=60),
 }
 
+# MCP v2 服务配置。生产环境应在 config/conf.py 中覆盖监听地址和 Host/Origin 白名单。
+MCP_HOST = globals().get('MCP_HOST', '127.0.0.1')
+MCP_PORT = globals().get('MCP_PORT', 2260)
+MCP_PATH = globals().get('MCP_PATH', '/mcp')
+MCP_ALLOWED_HOSTS = globals().get(
+    'MCP_ALLOWED_HOSTS',
+    ['127.0.0.1', '127.0.0.1:*', 'localhost', 'localhost:*'],
+)
+MCP_ALLOWED_ORIGINS = globals().get('MCP_ALLOWED_ORIGINS', [])
+MCP_MAX_REQUEST_BODY_SIZE = globals().get(
+    'MCP_MAX_REQUEST_BODY_SIZE', 1024 * 1024
+)
+MCP_MAX_RESULT_BYTES = globals().get('MCP_MAX_RESULT_BYTES', 512 * 1024)
+
 # 跨域配置/可用nginx处理,无需引入corsheaders
 CORS_ORIGIN_ALLOW_ALL = True
 CORS_ALLOW_CREDENTIALS = True