MCP server

@server.tool(
    description=(
        'Add the premises a conclusion rests on. Each record is {"conclusion": <id of a node of a '
        'type the rubric\'s premises apply to>, "premises": [{"stated_at": <chunk id>} or '
        '{"statement": <a premise the text leaves unsaid, in words>}]}. A stated premise links the '
        "chunk that states it; an unsaid one becomes an analyst assumption. The result lists the nodes "
        "`applied`, the records `refused`, each with a `fix`, and the `overflow`: unsaid premises not "
        "added because their conclusion already has as many as the rubric allows."
    )
)
def add_premises(
    run: Run = None,
    records: Annotated[Any, Field(description="A list of premise records.")] = None,
    model: Model = None,
    agent: Agent = None,
    note: Note = None,
) -> dict[str, Any]:
    """Add premises."""
    return _answer(lambda: tools.add_premises(_run(run), records, model=model, agent=agent, note=note))

@server.tool(
    description=(
        'Type, split and link chunks. Each record is {"node": <chunk id>, "type": <document type>, '
        '"form": {<that type\'s fields>}, "rests_on": [<ids>], "hedge": <phrase>}, with `form` '
        "only for a type that has fields (every required one given) and `rests_on` only for a type "
        "that rests on something, naming nodes of the types it may rest on; or, for a chunk that "
        'mixes kinds of statement, a split {"node": <chunk id>, "split": [<clauses copied verbatim '
        'from its text>], "children": [<one record without node per clause>]}, whose clauses become '
        "nodes s-2a, s-2b, ... in order. `node` is a chunk id exactly as start_run listed it, or the id "
        "of a clause a split has created; never an id with its type. A value of a form field must be "
        "written in the node's text. A node may be typed again until a check targets it; a split chunk "
        "is not split again (type its clauses). The result lists the nodes `applied`, the records "
        "`refused`, each with a `fix` to apply before calling again, and the `unmapped_hedges`, hedge "
        "phrases the rubric gives no credence for."
    )
)
def classify(
    run: Run = None,
    records: Annotated[Any, Field(description="A list of classify records.")] = None,
    model: Model = None,
    agent: Agent = None,
    note: Note = None,
) -> dict[str, Any]:
    """Type, split and link chunks."""
    return _answer(lambda: tools.classify(_run(run), records, model=model, agent=agent, note=note))

@server.tool(
    description=(
        "Finish the run: derive every verdict from the checks, run the core's check of the graph, and "
        "write graph.json, rejections.json and report.json. Returns the core's findings; fix them and "
        "finish again. Any later call marks the run unfinished until the next finish_run."
    )
)
def finish_run(run: Run = None, model: Model = None, agent: Agent = None, note: Note = None) -> dict[str, Any]:
    """Finish the run."""
    return _answer(lambda: tools.finish_run(_run(run), model=model, agent=agent, note=note))

@server.tool(
    description=(
        "Read the run as it stands: every node's id, type and text, with its verdict or check result "
        "where it has one; the relations; every refused record so far, with the call that sent it; "
        "how many calls the run has; and whether it is finished."
    )
)
def get_state(run: Run = None) -> dict[str, Any]:
    """Read the run."""
    return _answer(lambda: session.get_state(_run(run)))

@server.tool(
    description=(
        "Record a citation check: whether a passage of the cited work supports the citing node. "
        "`reference` is the bibliography entry's node id (r-N), not its key. `source` is the file of "
        "the run's cited/ copy (its name) or an http(s) URL; the tool reads it itself, so never paste "
        "its text. `passage` is quoted verbatim from the source. The check passes or fails as "
        "`supports` says only when the passage is in the source and the source is tied to the "
        "reference: the cited/ file the entry's key names, unchanged since start_run; the arXiv paper "
        "the entry names (give its arxiv.org URL); or the page the entry's DOI resolves to (give the "
        "doi.org URL). Any other source gives an error check, which moves no credence. A target whose "
        "citation check passed or failed takes no other; after one that errs, another may be made."
    )
)
def judge_citation(  # noqa: PLR0913, PLR0917 - the tool's arguments, each a separate choice
    run: Run = None,
    target: Annotated[Any, Field(description="The citing node's id.")] = None,
    reference: Annotated[Any, Field(description="The bibliography entry's node id, r-N.")] = None,
    source: Annotated[Any, Field(description="A file of the run's cited/ copy, or an http(s) URL.")] = None,
    passage: Annotated[Any, Field(description="The passage of the source, verbatim.")] = None,
    supports: Annotated[Any, Field(description="Whether the passage supports the citing node.")] = None,
    model: Model = None,
    agent: Agent = None,
    note: Note = None,
) -> dict[str, Any]:
    """Record a citation check."""
    return _answer(
        lambda: tools.judge_citation(
            _run(run),
            target,
            reference,
            source,
            passage,
            supports=supports,
            model=model,
            agent=agent,
            note=note,
        )
    )

def refuse_invented(name: Any, arguments: Mapping[str, Any]) -> None:
    """Refuse a call to a tool there is not, or one that sends an argument the tool does not take.

    The protocol would answer an unknown tool with an error of its own and drop an unknown argument
    silently, so a model's typo would go unnoticed; the server and the ``tool`` command both refuse
    them instead, naming the tools there are or the arguments the tool takes.

    Args:
        name: The tool.
        arguments: The call's arguments.

    Raises:
        RunError: If there is no such tool, or an argument is one the tool does not take.
    """
    if not isinstance(name, str) or name not in TOOLS:
        raise RunError("unknown-tool", f"there is no tool {name!r}; the tools are {list(TOOLS)}")
    invented = sorted(set(arguments) - {"run", *taken(name)})
    if invented:
        raise RunError("invalid-argument", f"{name} takes no {invented}; it takes {['run', *taken(name)]}")

@server.tool(
    description=(
        "Run arithmetic and maths checks on nodes of a type the rubric checks. A proposal is "
        '{"target": <id>, "type": "recompute-numbers", "nodes": [<chunk ids>], "expression": '
        '<arithmetic>, "expected": <the reported value as written>} or {"target": <id>, "type": '
        '"rederive-maths", "nodes": [<equation chunk ids>], "lhs": <expression>, "rhs": '
        '<expression>, "symbols": [{"name", "assumptions": [...], "domain": [low, high]}], '
        '"assumption": <in words>}. Every number in a recompute-numbers expression, the expected '
        "value included, must be written in the cited chunks: no constant they do not write (write "
        "a - a * p, not a * (1 - p)). Checks are final: a check is never withdrawn or run again on its "
        "target, nor run on another target as another encoding of the same step, and a failed check "
        "stays. Citations go to judge_citation. Refused proposals come back with a `fix`."
    )
)
def run_checks(
    run: Run = None,
    proposals: Annotated[Any, Field(description="A list of check proposals.")] = None,
    model: Model = None,
    agent: Agent = None,
    note: Note = None,
) -> dict[str, Any]:
    """Run maths checks."""
    return _answer(lambda: tools.run_checks(_run(run), proposals, model=model, agent=agent, note=note))

def serve() -> None:
    """Serve the tools over stdio until the harness closes the connection."""
    server.run("stdio")

@server.tool(
    description=(
        "Start a run: check the rubric, ingest the LaTeX paper, copy the cited works into the run, and "
        "return the chunks (id, unit, type, text, label or bibliography key), the document types with "
        "their form fields and what each rests on, the hedge phrases, and the check types with their "
        "targets. `run` must not exist or be empty. `cited` is a directory of the cited works, each a "
        "file named by its bibliography key (.txt, .tex or .pdf); citation checks read only the run's "
        "copy of it. `harness` names the agent harness."
    )
)
def start_run(  # noqa: PLR0913, PLR0917 - the tool's arguments, each a separate choice
    source: Annotated[Any, Field(description="The paper's main .tex file.")] = None,
    rubric: Annotated[Any, Field(description="The rubric file (TOML).")] = None,
    run: Run = None,
    harness: Annotated[Any, Field(description="The agent harness, as it reports itself.")] = None,
    model: Model = None,
    agent: Agent = None,
    note: Note = None,
    document: Annotated[Any, Field(description="An id for the document; the source's file name by default.")] = None,
    cited: Annotated[Any, Field(description="The directory of cited works, named by bibliography key.")] = None,
    store_source: Annotated[Any, Field(description="Keep the paper and fetched sources in the run.")] = False,
) -> dict[str, Any]:
    """Start a run."""

    def call() -> dict[str, Any]:
        if not isinstance(store_source, bool):
            raise RunError("invalid-argument", "store_source must be true or false")
        return session.start_run(
            _path("source", source),
            _path("rubric", rubric),
            _path("run", run),  # start_run makes the run
            harness=harness,
            model=model,
            agent=agent,
            note=note,
            document=document,
            cited=_optional("cited", cited),
            store_source=store_source,
        )

    return _answer(call)

def taken(name: str) -> list[str]:
    """The arguments a tool takes, besides the run directory, in order.

    Args:
        name: The tool.

    Returns:
        Their names.
    """
    return [parameter for parameter in inspect.signature(TOOLS[name]).parameters if parameter != "run"]