jobbsok/scripts/jobbsok_tools.py

#!/usr/bin/env python3
"""The `jobbsok-tools` host MCP server (plan Step 12).

Cowork and the Claude Code CLI have to reach the same logic (operator
decision 7). The CLI reaches it by importing `scripts/`; Cowork reaches it
through a plugin-declared stdio MCP server, which is this file.

Three decisions are worth stating, because each of them is a place a later
hand would reasonably do the other thing:

**Standard library only, protocol written out by hand.** The official Python
SDK would add a PyPI dependency the Cowork sandbox cannot install, and would
cost import time at every cold start against an MCP start-up timeout. Three
JSON-RPC methods is less code than the dependency it would replace.

**Nothing heavy is imported at module load.** `kandidat_schema`,
`vurdering`, `sak_status`, `dagens` and `beslutninger` are imported inside the
handlers, not at the top. The server has to answer `initialize` fast enough
that Cowork does not time it out, and it answers that one without touching any
of them.

**The M2 tools call the command line's own `main()`, with the streams
redirected** (Step 24). Parity between the two paths is not a nicety here: every
skill degrades to reading what the CLI produced when the server is absent, and
that instruction is only true if the content is the same. A second renderer on
this side would be a copy free to drift from the first, so what runs is the
same entry point -- same argument parsing, same output, same refusals -- and
the only thing this layer adds is turning an exit code into an MCP answer.

**Every tool takes an explicit `workspace`.** `jobbsok_lib.paths` refuses to
guess at a home directory, and this process runs unsandboxed on the operator's
machine (risk C6). A default here would put the guess back at exactly the
layer where a caller-supplied path meets the file system. `selvsjekk` takes it
too, and reports what it resolved to -- a self-check that could not tell you
which workspace it was looking at would be answering a smaller question than
the one asked.

Also usable from the command line, which is not decoration: the
`kandidatvurdering` skill's degradation branch tells the operator to run the
arithmetic from a terminal when the server is absent, and a CLI that did not
exist would make that instruction false.

    python3 scripts/jobbsok_tools.py                       # serve on stdio
    python3 scripts/jobbsok_tools.py --verktoy             # print tools/list
    python3 scripts/jobbsok_tools.py kandidat_valider \\
        --workspace ~/jobbsok-workspace                    # call one tool
"""

import json
import os
import sys

SERVER_NAME = "jobbsok-tools"
SERVER_VERSION = "0.1.0"
DEFAULT_PROTOCOL_VERSION = "2024-11-05"

#: JSON-RPC error codes this server emits. Method-not-found and invalid-params
#: are the standard ones; a tool that raised is reported as an MCP tool error
#: (`isError`) rather than a protocol error, because the call was well-formed.
INVALID_PARAMS = -32602
METHOD_NOT_FOUND = -32601

STANDARD_PROFIL = "profil/kandidat.md"

WORKSPACE_ARG = {
    "type": "string",
    "description": (
        "Absolutt sti til arbeidsområdet. Påkrevd. Det finnes ingen "
        "implisitt standard — serveren gjetter aldri på hjemmekatalogen."
    ),
}

#: The injected clock. Every M2 tool takes it, and none of them reads the wall
#: clock when it is absent only because a caller that cannot pin the date is
#: still better served by today than by a refusal -- but a `Verifiser`-step
#: that wants a reproducible answer passes it.
TODAY_ARG = {
    "type": "string",
    "description": (
        "Datoen som skal brukes, YYYY-MM-DD. Utelates den, brukes dagens "
        "dato. Oppgi den i alt som skal kunne reproduseres."
    ),
}

SAK_ARG = {
    "type": "string",
    "description": "Begrens til én sak-id. Utelates den, tas hele arbeidsområdet.",
}

TOOLS = [
    {
        "name": "kandidat_valider",
        "description": (
            "Valider kandidatprofilen mot kontrakten i build-brief 5.1 og "
            "returner rapporten som JSON: gyldig, feil, advarsler, seksjoner, "
            "vekter og vekter_kilde. Skriver ingenting."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "workspace": WORKSPACE_ARG,
                "sti": {
                    "type": "string",
                    "description": (
                        "Sti til profilen, relativt til arbeidsområdet. "
                        "Standard: %s" % STANDARD_PROFIL
                    ),
                },
            },
            "required": ["workspace"],
            "additionalProperties": False,
        },
    },
    {
        "name": "annonse_vurder",
        "description": (
            "Kjør hardfiltrene og regn ut den vektede scoren for en annonse "
            "mot kandidatprofilen. Returnerer score, delscore, vekter, "
            "vekt_hash, verdikt, avvisninger, advarsler og bekymringer. "
            "Avvisninger og advarsler er to utfall, ikke ett. Skriver "
            "ingenting."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "workspace": WORKSPACE_ARG,
                "annonse": {
                    "type": "object",
                    "description": "Frontmatteren fra annonsen, som et kart.",
                },
                "brodtekst": {
                    "type": "string",
                    "description": "Annonsens brødtekst, uten frontmatter.",
                },
                "delscore": {
                    "type": "object",
                    "description": (
                        "Modellens bidrag: ett heltall 0–100 per kriterium. "
                        "Alle kriteriene må være med."
                    ),
                },
                "bekymringer": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "Fritekst som følger med uendret.",
                },
                "profil_sti": {
                    "type": "string",
                    "description": (
                        "Sti til profilen, relativt til arbeidsområdet. "
                        "Standard: %s" % STANDARD_PROFIL
                    ),
                },
            },
            "required": ["workspace", "annonse", "brodtekst", "delscore"],
            "additionalProperties": False,
        },
    },
    {
        "name": "selvsjekk",
        "description": (
            "Rapporter hvilken tolk som faktisk serverer verktøyene, hvilken "
            "versjon av ingest-vakten som er installert, hvilket byggestempel "
            "plugin-en har, og hvilket arbeidsområde som ble resolvert. "
            "Brukes til å avsløre en hurtigbufret plugin-versjon."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {"workspace": WORKSPACE_ARG},
            "required": ["workspace"],
            "additionalProperties": False,
        },
    },
    {
        "name": "sak_status",
        "description": (
            "Utled status, ventende part, siste aktivitet, neste frist og "
            "stillhets-flagg for hver sak, fra logg.jsonl. sak.md er en "
            "hurtigbuffer og loggen vinner. Skriver ingenting."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "workspace": WORKSPACE_ARG,
                "today": TODAY_ARG,
                "sak": SAK_ARG,
                "format": {
                    "type": "string",
                    "enum": ["tekst", "json"],
                    "description": "Utdataform. Standard: tekst.",
                },
            },
            "required": ["workspace"],
            "additionalProperties": False,
        },
    },
    {
        "name": "sak_status_check",
        "description": (
            "Samme utledning, men rapporterer om en hurtigbuffer i sak.md er "
            "uenig med loggen. Avviket staar i teksten, med saken navngitt; "
            "kommandolinja svarer i tillegg med exit-kode 1, som MCP ikke har. "
            "Retter ingenting og skriver ingenting."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "workspace": WORKSPACE_ARG,
                "today": TODAY_ARG,
                "sak": SAK_ARG,
            },
            "required": ["workspace"],
            "additionalProperties": False,
        },
    },
    {
        "name": "dagens",
        "description": (
            "Dagens arbeidsbilde som ren tekst: hva som krever handling, hva "
            "som har gaatt stille, kommende frister og saker per tilstand. "
            "Byte-stabil for samme arbeidsomraade og samme dato. Skriver "
            "ingenting."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {"workspace": WORKSPACE_ARG, "today": TODAY_ARG},
            "required": ["workspace"],
            "additionalProperties": False,
        },
    },
    {
        "name": "beslutning_append",
        "description": (
            "Legg til én beslutning i den append-only beslutningsloggen "
            "(build-brief 5.4), eller en korrigering som navngir en tidligere "
            "beslutning. En retting er alltid en ny linje, aldri en "
            "redigering. Returnerer posten som ble skrevet."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "workspace": WORKSPACE_ARG,
                "post": {
                    "type": "object",
                    "description": (
                        "Posten. Feltene er build-brief 5.4: id, dato, kilde, "
                        "url, tittel, arbeidsgiver, beslutning, arsak, notat, "
                        "score_da, delscore, vekter, vekt_hash, korrigerer."
                    ),
                },
                "korriger": {
                    "type": "boolean",
                    "description": (
                        "Posten er en korrigering av en tidligere beslutning. "
                        "Standard: false."
                    ),
                },
            },
            "required": ["workspace", "post"],
            "additionalProperties": False,
        },
    },
]


class ToolError(Exception):
    """A tool refused or failed. Reported as an MCP tool error, not a protocol one."""


def log(message):
    """Diagnostics go to stderr; stdout carries protocol traffic only."""
    print("[%s] %s" % (SERVER_NAME, message), file=sys.stderr, flush=True)


def plugin_root():
    """The plugin root: what the host declared, else this file's parent."""
    declared = (os.environ.get("CLAUDE_PLUGIN_ROOT") or "").strip()
    if declared:
        return os.path.realpath(os.path.expanduser(declared))
    return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def til_json(payload):
    """The canonical JSON this server returns.

    `ensure_ascii=False` for the same reason `kandidat_schema.report_json`
    uses it: the payload quotes the operator's own words back, and an escaped
    o-slash is no longer the word that was written.
    """
    return json.dumps(payload, ensure_ascii=False, indent=2) + "\n"


def _workspace(arguments):
    from jobbsok_lib import paths

    raw = arguments.get("workspace")
    if not isinstance(raw, str) or not raw.strip():
        raise ToolError(
            "workspace mangler. Hvert verktøy tar et eksplisitt arbeidsområde; "
            "det finnes ingen standard, og serveren gjetter ikke."
        )
    try:
        return paths.workspace_root(raw)
    except paths.WorkspaceError as exc:
        raise ToolError(str(exc))


def _safe(root, relativ):
    from jobbsok_lib import paths

    try:
        return paths.safe_join(root, *relativ.split("/"))
    except paths.WorkspaceError as exc:
        raise ToolError(str(exc))


def tool_kandidat_valider(arguments):
    import kandidat_schema

    root = _workspace(arguments)
    relativ = arguments.get("sti") or STANDARD_PROFIL
    _safe(root, relativ)
    try:
        rapport = kandidat_schema.validate_file(root, *relativ.split("/"))
    except OSError as exc:
        raise ToolError("kunne ikke lese %s: %s" % (relativ, exc))
    return kandidat_schema.report_json(rapport)


def tool_annonse_vurder(arguments):
    import vurdering

    root = _workspace(arguments)
    relativ = arguments.get("profil_sti") or STANDARD_PROFIL
    _safe(root, relativ)

    annonse = arguments.get("annonse")
    if not isinstance(annonse, dict):
        raise ToolError("annonse mangler; forventet frontmatteren som et kart")
    brodtekst = arguments.get("brodtekst")
    if not isinstance(brodtekst, str):
        raise ToolError("brodtekst mangler; forventet annonsens brødtekst som tekst")

    try:
        profil = vurdering.les_profil_fil(root, *relativ.split("/"))
    except OSError as exc:
        raise ToolError("kunne ikke lese %s: %s" % (relativ, exc))

    payload = {
        "delscore": arguments.get("delscore"),
        "bekymringer": arguments.get("bekymringer") or [],
    }
    try:
        resultat = vurdering.vurder(profil, annonse, brodtekst, payload)
    except vurdering.DelscoreError as exc:
        raise ToolError(str(exc))
    return til_json(resultat)


def _streng(arguments, navn):
    """An optional string argument, refused rather than coerced.

    Types are checked before anything reaches `argparse`: a non-string would
    become a malformed argv, and argparse answers a malformed argv with
    SystemExit -- which is a BaseException and would take the server down
    rather than returning a tool error.
    """
    verdi = arguments.get(navn)
    if verdi is None:
        return None
    if not isinstance(verdi, str) or not verdi.strip():
        raise ToolError("%s må være en ikke-tom tekst, fikk %r" % (navn, verdi))
    return verdi


def _kjor_cli(modul, argv):
    """Run a command-line entry point in-process and return its stdout.

    The exit code is translated, not swallowed: a hard failure becomes a tool
    error carrying what the CLI wrote to stderr. `sak_status`'s exit code 1 --
    "a cache diverged" -- is deliberately not a failure; it is an answer, and
    it is in the text.
    """
    import io as _io

    ut, feil = _io.StringIO(), _io.StringIO()
    kode = modul.main(argv, stdout=ut, stderr=feil)
    if kode == modul.EXIT_FEIL:
        raise ToolError(feil.getvalue().strip() or "kommandoen feilet uten melding")
    return ut.getvalue()


def _sak_status_argv(arguments):
    argv = ["--workspace", _workspace(arguments)]
    today = _streng(arguments, "today")
    if today:
        argv.extend(["--today", today])
    sak = _streng(arguments, "sak")
    if sak:
        argv.extend(["--sak", sak])
    return argv


def tool_sak_status(arguments):
    import sak_status

    argv = _sak_status_argv(arguments)
    form = _streng(arguments, "format") or "tekst"
    if form not in ("tekst", "json"):
        raise ToolError("format må være tekst eller json, fikk %r" % (form,))
    return _kjor_cli(sak_status, argv + ["--format", form])


def tool_sak_status_check(arguments):
    import sak_status

    # Text, not JSON: the CLI appends the divergence line to stdout after the
    # report, which is natural in the text form and would make the JSON form
    # unparseable. The answer is in that line, because MCP has no exit code.
    return _kjor_cli(sak_status, _sak_status_argv(arguments) + ["--check"])


def tool_dagens(arguments):
    import dagens

    argv = ["--workspace", _workspace(arguments)]
    today = _streng(arguments, "today")
    if today:
        argv.extend(["--today", today])
    return _kjor_cli(dagens, argv)


def tool_beslutning_append(arguments):
    import beslutninger

    root = _workspace(arguments)
    post = arguments.get("post")
    if not isinstance(post, dict):
        raise ToolError(
            "post mangler; forventet beslutningen som et kart med feltene fra "
            "build-brief 5.4"
        )
    argv = ["--workspace", root, "--json", json.dumps(post, ensure_ascii=False)]
    if arguments.get("korriger"):
        argv.append("--korriger")
    return _kjor_cli(beslutninger, argv)


def tool_selvsjekk(arguments):
    root = _workspace(arguments)

    try:
        import llm_ingestion_guard

        guard = getattr(llm_ingestion_guard, "__version__", None)
    except ImportError:
        guard = None

    stamp_sti = os.path.join(plugin_root(), "BUILD_STAMP")
    try:
        with open(stamp_sti, "r", encoding="utf-8") as handle:
            stamp = handle.read().strip() or None
    except OSError:
        # Absent is a fact, and it is reported as one. A self-check that
        # silently omitted the stamp would let a cached build pass unnoticed,
        # which is the whole reason the stamp exists (risk H11).
        stamp = None

    return til_json(
        {
            "server": SERVER_NAME,
            "server_versjon": SERVER_VERSION,
            "python_executable": sys.executable,
            "python_version": sys.version.split()[0],
            "python_version_info": list(sys.version_info[:3]),
            "guard_versjon": guard,
            "plugin_root": plugin_root(),
            "build_stamp": stamp,
            "workspace": root,
        }
    )


#: Insertion order is the order `tools/list` publishes, and the golden file
#: pins it. Keep it the same as TOOLS above.
HANDLERS = {
    "kandidat_valider": tool_kandidat_valider,
    "annonse_vurder": tool_annonse_vurder,
    "selvsjekk": tool_selvsjekk,
    "sak_status": tool_sak_status,
    "sak_status_check": tool_sak_status_check,
    "dagens": tool_dagens,
    "beslutning_append": tool_beslutning_append,
}


def call_tool(name, arguments):
    """Run one tool and return its text, or raise :class:`ToolError`."""
    handler = HANDLERS.get(name)
    if handler is None:
        raise ToolError("ukjent verktøy: %r" % (name,))
    return handler(arguments or {})


def handle(request):
    """Return a response dict, or None for a notification (no reply expected)."""
    method = request.get("method")
    req_id = request.get("id")

    if req_id is None:
        return None

    if method == "initialize":
        params = request.get("params") or {}
        # Echo the client's protocol version when it offers one; announcing a
        # newer version than the client speaks is how handshakes fail silently.
        result = {
            "protocolVersion": params.get("protocolVersion") or DEFAULT_PROTOCOL_VERSION,
            "capabilities": {"tools": {}},
            "serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION},
        }
    elif method == "tools/list":
        result = {"tools": TOOLS}
    elif method == "tools/call":
        params = request.get("params") or {}
        name = params.get("name")
        if name not in HANDLERS:
            return _error(req_id, INVALID_PARAMS, "ukjent verktøy: %r" % (name,))
        try:
            text = call_tool(name, params.get("arguments"))
        except ToolError as exc:
            result = {"content": [{"type": "text", "text": str(exc)}], "isError": True}
        except Exception as exc:  # noqa: BLE001 -- a crash must not kill the server
            log("%s raised %s: %s" % (name, type(exc).__name__, exc))
            result = {
                "content": [
                    {"type": "text", "text": "%s: %s" % (type(exc).__name__, exc)}
                ],
                "isError": True,
            }
        else:
            result = {"content": [{"type": "text", "text": text}], "isError": False}
    elif method == "ping":
        result = {}
    else:
        return _error(req_id, METHOD_NOT_FOUND, "ukjent metode: %r" % (method,))

    return {"jsonrpc": "2.0", "id": req_id, "result": result}


def _error(req_id, code, message):
    return {"jsonrpc": "2.0", "id": req_id, "error": {"code": code, "message": message}}


def serve(stdin=None, stdout=None):
    """The stdio loop: newline-delimited JSON-RPC, nothing else on stdout."""
    stdin = sys.stdin if stdin is None else stdin
    stdout = sys.stdout if stdout is None else stdout
    for line in stdin:
        line = line.strip()
        if not line:
            continue
        try:
            request = json.loads(line)
        except ValueError as exc:
            log("undecodable line: %s" % exc)
            continue
        response = handle(request)
        if response is not None:
            stdout.write(json.dumps(response, ensure_ascii=False) + "\n")
            stdout.flush()
    return 0


def main(argv=None):
    argv = list(sys.argv[1:] if argv is None else argv)

    if not argv:
        log("serving on %s" % sys.version.split()[0])
        return serve()

    if argv[0] in ("-h", "--help"):
        sys.stdout.write(__doc__)
        return 0

    if argv[0] == "--verktoy":
        sys.stdout.write(til_json({"tools": TOOLS}))
        return 0

    name = argv[0]
    arguments = {}
    rest = argv[1:]
    while rest:
        flagg = rest.pop(0)
        if not flagg.startswith("--"):
            print("jobbsok-tools: forventet et --flagg, fikk %r" % flagg, file=sys.stderr)
            return 2
        if not rest:
            print("jobbsok-tools: %s mangler en verdi" % flagg, file=sys.stderr)
            return 2
        verdi = rest.pop(0)
        nokkel = flagg[2:]
        # `--json` carries the arguments a shell cannot express: the listing
        # mapping and the sub-score table. Scalars stay scalars.
        if nokkel == "json":
            arguments.update(json.loads(verdi))
        else:
            arguments[nokkel] = verdi

    try:
        sys.stdout.write(call_tool(name, arguments))
    except ToolError as exc:
        print("jobbsok-tools: %s" % exc, file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())