"""tests/paper/test_mcp_protocol.py —— MCP 协议层测试(list_tools / call_tool 往返)。 验证:9 个工具正确暴露、call_tool 返回可解析的 JSON(structuredContent 或 content 文本块)。 镜像 bioflow_service/tests/test_mcp_server.py 的 call_tool 集成模式。 """ from __future__ import annotations import json from conftest import _run from paper_writer_service.app.mcp import paper_mcp EXPECTED_TOOLS = { "propose_paper_outline", "write_paper_section", "get_paper_progress", "search_refs", "revise_paper_section", "revise_paper", "get_paper", "complete_paper", "cancel_paper", } def _content_text(result): """从 CallToolResult 抽取文本内容(无 structuredContent 时)。""" for block in getattr(result, "content", None) or []: text = getattr(block, "text", None) if text: return json.loads(text) return {} def _call(name, args): r = _run(paper_mcp.call_tool(name, args)) return getattr(r, "structuredContent", None) or _content_text(r) def test_list_tools_exposes_nine(): tools = _run(paper_mcp.list_tools()) names = {getattr(t, "name", None) for t in tools} assert EXPECTED_TOOLS == names def test_call_tool_propose_outline_roundtrip(paper_db, fake_llm, fake_search): fake_llm.responses.append({ "title": "水稻SPL基因家族研究进展", "abstract_expect": "综述", "sections": [{"section_id": "s1", "title": "引言", "key_points": [], "target_words": 1500}], }) r = _call("propose_paper_outline", {"topic": "水稻SPL基因家族", "paper_type": "综述"}) assert r.get("ok") is True assert r["paper_id"] assert r["title"] == "水稻SPL基因家族研究进展" def test_call_tool_not_found(paper_db): r = _call("get_paper", {"paper_id": "nope"}) assert r.get("ok") is False assert r.get("code") == "NOT_FOUND" def test_call_tool_content_is_json_string(paper_db, fake_llm, fake_search): """验证 content[0].text 是合法 JSON 字符串(BFF 裸 JSON-RPC 客户端解析面)。""" fake_llm.responses.append({ "title": "T", "abstract_expect": "A", "sections": [{"section_id": "s1", "title": "引言", "key_points": [], "target_words": 1500}], }) raw = _run(paper_mcp.call_tool("propose_paper_outline", {"topic": "水稻SPL"})) blocks = getattr(raw, "content", None) or [] assert blocks, "CallToolResult 无 content 块" text = getattr(blocks[0], "text", None) assert text is not None payload = json.loads(text) # 应为合法 JSON 字符串 assert payload.get("ok") is True