"""Sentence tools (语句): the smallest unit the paper model stores. One row is one sentence, and nothing about the writing path ever re-splits it — the writer decides where a sentence ends (see ``docs/OVERVIEW.md``). That makes these the tools to reach for when a paragraph is *almost* right: rewriting the whole paragraph to fix one line would relabel every sentence id in it, and an id is what a citation points at. :func:`sentence_search` reads across papers instead of within one, because the job it serves is consistency: a model that has already written "flood damage" in one paper should not start writing "GUL" in the next. """ from __future__ import annotations from typing import Annotated, Any from mcp.server.mcpserver import MCPServer from pydantic import Field from app.crud import paper as paper_crud from app.mcp import support as s from app.mcp.specs import CitationSpec from app.mcp.tools.registry import mcp_tool from app.models import Paper from app.schemas.paper import CitationInput, SentenceCreate, SentenceUpdate def register(server: MCPServer) -> None: """Register every sentence tool on ``server``.""" @mcp_tool(server, title="列出句子") def sentence_list( paper_id: Annotated[int, Field(description="论文 id")], position: Annotated[ int | None, Field(description="只看某一段；不传则返回全文句子") ] = None, heading: Annotated[ str | None, Field(description="只看某个标题下的句子") ] = None, include_empty: Annotated[ bool, Field(description="是否包含空句子（只挂引用的行）") ] = False, ) -> str: """列出论文里的句子，带句子 id、所属段落和引用。""" with s.session() as db: paper = s.paper_or_fail(db, paper_id) target = ( s.resolve_position(paper, position=position, heading=heading) if (position is not None or heading is not None) else None ) rows = [ row for row in paper.sentences if (target is None or row.paper_template_filed_sort == target) and (include_empty or row.content) ] rows.sort(key=lambda row: (row.paper_template_filed_sort, row.sort, row.id)) return s.dumps( { "paper_id": paper.id, "paper_title": paper.title, "total": len(rows), "items": [ { **s.sentence_row(row), "heading": _heading_at(paper, row.paper_template_filed_sort), } for row in rows ], } ) @mcp_tool(server, title="加一句") def sentence_add( paper_id: Annotated[int, Field(description="论文 id")], content: Annotated[str, Field(description="句子正文；会折叠成一行")], position: Annotated[ int | None, Field(description="加到哪一段；与 heading 二选一") ] = None, heading: Annotated[str | None, Field(description="加到哪个标题下")] = None, citations: Annotated[ list[CitationSpec] | None, Field(description="这句引用了哪些文献") ] = None, sort: Annotated[ int | None, Field(description="段内位置；不传则排在段尾") ] = None, ) -> str: """在某一段末尾追加一句话，可带引用。""" with s.session() as db: paper = s.paper_or_fail(db, paper_id) field_sort = s.resolve_position(paper, position=position, heading=heading) sentence = paper_crud.append_sentence( db, paper, SentenceCreate( paper_template_filed_sort=field_sort, content=content, citations=_citations(citations or []), sort=sort, ), ) return s.dumps( { "paper_id": paper.id, "position": field_sort, "heading": _heading_at(paper, field_sort), "sentence": s.sentence_row(sentence), } ) @mcp_tool(server, title="改一句") def sentence_update( paper_id: Annotated[int, Field(description="论文 id")], sentence_id: Annotated[int, Field(description="句子 id（见 sentence_list）")], content: Annotated[str | None, Field(description="新正文；不传则不改")] = None, position: Annotated[ int | None, Field(description="把这句话移到另一段；不传则不动") ] = None, sort: Annotated[ int | None, Field(description="段内排序值；不传则不动") ] = None, citations: Annotated[ list[CitationSpec] | None, Field(description="引用整组替换；传 [] 表示清空引用"), ] = None, ) -> str: """修改一句话的正文、所属段落、排序或引用（引用是整体替换）。""" if content is None and position is None and sort is None and citations is None: s.fail("没有给出任何要修改的内容") with s.session() as db: paper = s.paper_or_fail(db, paper_id) sentence = s.sentence_or_fail(paper, sentence_id) updated = paper_crud.update_sentence( db, paper, sentence, SentenceUpdate( content=content, paper_template_filed_sort=position, sort=sort, citations=None if citations is None else _citations(citations), ), ) return s.dumps( { "paper_id": paper.id, "sentence": s.sentence_row(updated), } ) @mcp_tool(server, title="删一句") def sentence_delete( paper_id: Annotated[int, Field(description="论文 id")], sentence_id: Annotated[int, Field(description="句子 id")], ) -> str: """删除一句话，连同它挂着的引用。""" with s.session() as db: paper = s.paper_or_fail(db, paper_id) sentence = s.sentence_or_fail(paper, sentence_id) snapshot = s.sentence_row(sentence) paper_crud.delete_sentence(db, paper, sentence) return s.dumps( { "paper_id": paper.id, "deleted": snapshot, } ) @mcp_tool(server, title="搜索句子") def sentence_search( keyword: Annotated[str, Field(description="要搜索的字词")], paper_id: Annotated[ int | None, Field(description="只在这篇论文里搜；不传则全库搜") ] = None, limit: Annotated[int, Field(description="最多返回多少条")] = 30, ) -> str: """跨论文搜索已经写过的句子，用来保持术语和说法一致。""" if not keyword.strip(): s.fail("keyword 不能为空") with s.session() as db: rows = paper_crud.search_sentences( db, keyword=keyword, paper_id=paper_id, limit=max(1, min(int(limit), 200)), ) return s.dumps( { "keyword": keyword, "total": len(rows), "items": [ { "paper_id": sentence.paper_id, "paper_title": title, "sentence_id": sentence.id, "position": sentence.paper_template_filed_sort, "content": sentence.content, } for sentence, title in rows ], } ) # --- helpers ----------------------------------------------------------------- def _citations(items: list[Any]) -> list[CitationInput]: """Convert the tool-level citation spec into the project's own payload.""" return [ CitationInput(quote=item.quote, reference_id=item.reference_id) for item in items ] def _heading_at(paper: Paper, field_sort: int) -> str | None: """What the paper's template calls this position, if it names it at all.""" for item in s.heading_index(paper): if item["position"] == field_sort: return str(item["heading"]) return None