MCP 工具

src.retrieval.cg_rag_mcp 将 CG_RAG 服务包装为 FastMCP 工具。工具返回统一结构，区分 MCP 调用是否成功以及底层 CG_RAG 操作是否成功。

工具概览

• cg_health：返回 CG_RAG 服务健康状态。
• cg_list_profiles：列出可用检索 profile。
• cg_retrieve_rerank：执行召回和重排，只返回候选条文。
• cg_constrained_generate：基于候选条文执行受约束生成，支持 generation_enable_thinking 单次覆盖。
• cg_rag：执行完整 CG_RAG 流程，支持 generation_enable_thinking 单次覆盖。

MCP 工具返回 JSON envelope；需要答案 token 流时使用 HTTP /cg-rag/rag/stream。

cg_rag_mcp

make_cg_rag_mcp_lifespan(get_mcp: Callable[[], FastMCP | None]) -> Callable[[Any], AsyncIterator[None]]

源代码位于： src/retrieval/cg_rag_mcp.py

def make_cg_rag_mcp_lifespan(get_mcp: Callable[[], FastMCP | None]) -> Callable[[Any], AsyncIterator[None]]:
    @asynccontextmanager
    async def lifespan(_app: Any) -> AsyncIterator[None]:
        mcp = get_mcp()
        if mcp is None:
            yield
            return
        async with mcp.session_manager.run():
            yield

    return lifespan

build_cg_rag_mcp(service: Any) -> FastMCP

源代码位于： src/retrieval/cg_rag_mcp.py

def build_cg_rag_mcp(service: Any) -> FastMCP:
    mcp = FastMCP("CG_RAG", json_response=True, streamable_http_path="/")

    @mcp.tool()
    async def cg_health() -> dict[str, Any]:
        """返回检索服务健康状态、active scope、可用 profile、rerank 和 LLM 配置状态。"""
        return await _call_service(lambda _cancel_token: service.health())

    @mcp.tool()
    async def cg_list_profiles() -> dict[str, Any]:
        """列出检索服务可用检索范围 profile，例如 usual、usual_plus_law 和 full。"""
        return await _call_service(lambda _cancel_token: {"profiles": service.profiles()})

    @mcp.tool()
    async def cg_retrieve_rerank(
        query: TextInput = None, scope: TextInput = "usual", topk: IntInput = 78
    ) -> dict[str, Any]:
        """执行检索服务召回和重排，只返回候选条文，不调用 LLM 生成。"""
        return await _call_service(
            lambda cancel_token: _call_with_optional_cancel(
                service.retrieve_rerank,
                CgRetrieveRerankRequest(query=query, scope=scope, topk=topk),
                cancel_token=cancel_token,
            )
        )

    @mcp.tool()
    async def cg_constrained_generate(
        query: TextInput = None,
        retrieval_docs: DocsInput = None,
        max_items: IntInput = 10,
        include_debug: BoolInput = False,
        generation_enable_thinking: BoolInput = None,
    ) -> dict[str, Any]:
        """基于调用方传入的候选条文执行受约束生成，返回候选索引和格式化条文。"""
        return await _call_service(
            lambda cancel_token: _call_with_optional_cancel(
                service.constrained_generate,
                CgConstrainedGenerateRequest(
                    query=query,
                    retrieval_docs=retrieval_docs,
                    max_items=max_items,
                    include_debug=include_debug,
                    generation_enable_thinking=generation_enable_thinking,
                ),
                cancel_token=cancel_token,
            )
        )

    @mcp.tool()
    async def cg_rag(
        query: TextInput = None,
        scope: TextInput = "usual",
        topk: IntInput = 78,
        max_items: IntInput = 10,
        include_debug: BoolInput = False,
        generation_enable_thinking: BoolInput = None,
    ) -> dict[str, Any]:
        """执行完整检索增强流程：召回、重排、受约束生成。该流程不包含 domain gate。"""
        return await _call_service(
            lambda cancel_token: _call_with_optional_cancel(
                service.rag,
                CgRagRequest(
                    query=query,
                    scope=scope,
                    topk=topk,
                    max_items=max_items,
                    include_debug=include_debug,
                    generation_enable_thinking=generation_enable_thinking,
                ),
                cancel_token=cancel_token,
            ),
            operation_ok_from_data=_cg_rag_operation_ok,
        )

    @mcp.tool()
    async def cg_vision_answer(
        images: ImagesInput = None,
        question: TextInput = "请检测图中的安全隐患",
        enable_thinking: BoolInput = None,
    ) -> dict[str, Any]:
        """自由生成：直接根据施工现场图片识别安全隐患并给出整改建议，不检索法规，回答不带引用。"""
        return await _call_service(
            lambda cancel_token: _call_with_optional_cancel(
                service.vision_answer,
                CgVisionAnswerRequest(
                    images=images,
                    question=question,
                    enable_thinking=enable_thinking,
                ),
                cancel_token=cancel_token,
            )
        )

    @mcp.tool()
    async def cg_vision_observe(
        images: ImagesInput = None,
        question: TextInput = "请检测图中的安全隐患",
    ) -> dict[str, Any]:
        """结构化视觉观察：返回校验过的隐患标签与展示文本，供调用方据此构造检索问题。"""
        return await _call_service(
            lambda cancel_token: _call_with_optional_cancel(
                service.vision_observe,
                CgVisionAnswerRequest(images=images, question=question),
                cancel_token=cancel_token,
            )
        )

    @mcp.tool()
    async def cg_vision_agentic(
        images: ImagesInput = None,
        question: TextInput = "请检测图中的安全隐患",
        scope: TextInput = "full",
        max_search_rounds: IntInput = None,
        topk: IntInput = None,
        enable_thinking: BoolInput = None,
    ) -> dict[str, Any]:
        """智能生成：模型先观察图片，再自行检索法规条文，最后输出带 [cite^n] 引用的隐患判定。"""
        return await _call_service(
            lambda cancel_token: _call_with_optional_cancel(
                service.vision_agentic,
                CgVisionAgenticRequest(
                    images=images,
                    question=question,
                    scope=scope,
                    max_search_rounds=max_search_rounds,
                    topk=topk,
                    enable_thinking=enable_thinking,
                ),
                cancel_token=cancel_token,
            )
        )

    return mcp