feat(m1): add jobbsok-tools stdio mcp server and launcher

This commit is contained in:
Kjell Tore Guttormsen 2026-09-05 06:48:55 +02:00
commit 89bcd1b038
6 changed files with 1057 additions and 0 deletions

8
.mcp.json Normal file
View file

@ -0,0 +1,8 @@
{
"mcpServers": {
"jobbsok-tools": {
"command": "bash",
"args": ["${CLAUDE_PLUGIN_ROOT}/scripts/jobbsok_tools_launch.sh"]
}
}
}

409
scripts/jobbsok_tools.py Normal file
View file

@ -0,0 +1,409 @@
#!/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` and
`vurdering` 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 either module.
**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."
),
}
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,
},
},
]
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 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,
}
)
HANDLERS = {
"kandidat_valider": tool_kandidat_valider,
"annonse_vurder": tool_annonse_vurder,
"selvsjekk": tool_selvsjekk,
}
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())

107
scripts/jobbsok_tools_launch.sh Executable file
View file

@ -0,0 +1,107 @@
#!/bin/bash
# Start the jobbsok-tools stdio MCP server on an interpreter that is new enough.
#
# bash 3.2-clean and ASCII-only on purpose: the system bash on this Mac is 3.2,
# and a multibyte character has crashed a `set -u` script here before.
#
# Why this file exists at all, rather than .mcp.json naming python3 directly:
# a GUI-spawned process on this Mac can resolve python3 to 3.9.6 under an
# empty launchd PATH (risk H2). A server running on 3.9.6 is worse than no
# server, because the degradation branch already covers an absent server and
# nothing covers a silently wrong one. So every candidate below is gated on
# the version, and when none passes the launcher refuses instead of falling
# through to whatever python3 the PATH happens to offer.
#
# Resolution order, first candidate that passes the gate wins:
# 1. $JOBBSOK_PYTHON -- the manual override. Named explicitly by
# the operator, so a version failure here
# is refused outright rather than skipped:
# falling back from an interpreter someone
# asked for would hide their mistake.
# 2. $CLAUDE_PLUGIN_DATA/venv -- what scripts/bootstrap.sh builds against
# an installed plugin. This is the
# supported route in an installed copy.
# 3. $CLAUDE_PLUGIN_ROOT/.venv -- the development environment. The packaged
# archive excludes the virtualenv, so this
# branch never fires in an installed copy.
# 4. python3 on PATH -- last resort, and only at 3.10 or newer.
set -eu
MIN_MAJOR=3
MIN_MINOR=10
SCRIPT_DIR=$(cd "$(dirname "$0")" && pwd)
if [ -n "${CLAUDE_PLUGIN_ROOT:-}" ]; then
PLUGIN_ROOT="$CLAUDE_PLUGIN_ROOT"
else
PLUGIN_ROOT=$(cd "$SCRIPT_DIR/.." && pwd)
fi
SERVER="${PLUGIN_ROOT}/scripts/jobbsok_tools.py"
version_ok() {
"$1" -c "import sys; sys.exit(0 if sys.version_info >= ($MIN_MAJOR, $MIN_MINOR) else 1)" \
>/dev/null 2>&1
}
version_of() {
"$1" -c "import sys; print('%d.%d.%d' % sys.version_info[:3])" 2>/dev/null || echo "ukjent"
}
usable() {
[ -n "$1" ] || return 1
command -v "$1" >/dev/null 2>&1 || return 1
version_ok "$1"
}
PYTHON_BIN=""
if [ -n "${JOBBSOK_PYTHON:-}" ]; then
if ! command -v "$JOBBSOK_PYTHON" >/dev/null 2>&1; then
echo "jobbsok-tools: JOBBSOK_PYTHON=$JOBBSOK_PYTHON is not executable." >&2
echo "jobbsok-tools: refusing to start. Point it at a Python ${MIN_MAJOR}.${MIN_MINOR}+ interpreter." >&2
exit 1
fi
if ! version_ok "$JOBBSOK_PYTHON"; then
echo "jobbsok-tools: JOBBSOK_PYTHON=$JOBBSOK_PYTHON is Python $(version_of "$JOBBSOK_PYTHON")." >&2
echo "jobbsok-tools: ${MIN_MAJOR}.${MIN_MINOR}+ is required; refusing to start on it." >&2
exit 1
fi
PYTHON_BIN="$JOBBSOK_PYTHON"
fi
if [ -z "$PYTHON_BIN" ] && [ -n "${CLAUDE_PLUGIN_DATA:-}" ]; then
CANDIDATE="${CLAUDE_PLUGIN_DATA}/venv/bin/python"
if usable "$CANDIDATE"; then
PYTHON_BIN="$CANDIDATE"
fi
fi
if [ -z "$PYTHON_BIN" ]; then
CANDIDATE="${PLUGIN_ROOT}/.venv/bin/python3"
if usable "$CANDIDATE"; then
PYTHON_BIN="$CANDIDATE"
fi
fi
if [ -z "$PYTHON_BIN" ]; then
CANDIDATE=$(command -v python3 2>/dev/null || true)
if usable "$CANDIDATE"; then
PYTHON_BIN="$CANDIDATE"
fi
fi
if [ -z "$PYTHON_BIN" ]; then
echo "jobbsok-tools: found no Python ${MIN_MAJOR}.${MIN_MINOR}+ interpreter." >&2
echo "jobbsok-tools: tried JOBBSOK_PYTHON, \$CLAUDE_PLUGIN_DATA/venv, \$CLAUDE_PLUGIN_ROOT/.venv and python3 on PATH." >&2
echo "jobbsok-tools: run 'bash scripts/bootstrap.sh' against the installed plugin, or set JOBBSOK_PYTHON." >&2
echo "jobbsok-tools: refusing to start rather than serving on an interpreter below ${MIN_MAJOR}.${MIN_MINOR}." >&2
exit 1
fi
if [ ! -f "$SERVER" ]; then
echo "jobbsok-tools: server not found at $SERVER" >&2
exit 1
fi
exec "$PYTHON_BIN" "$SERVER" "$@"

View file

@ -0,0 +1,85 @@
{
"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": {
"type": "string",
"description": "Absolutt sti til arbeidsområdet. Påkrevd. Det finnes ingen implisitt standard — serveren gjetter aldri på hjemmekatalogen."
},
"sti": {
"type": "string",
"description": "Sti til profilen, relativt til arbeidsområdet. Standard: profil/kandidat.md"
}
},
"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": {
"type": "string",
"description": "Absolutt sti til arbeidsområdet. Påkrevd. Det finnes ingen implisitt standard — serveren gjetter aldri på hjemmekatalogen."
},
"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: profil/kandidat.md"
}
},
"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": {
"type": "string",
"description": "Absolutt sti til arbeidsområdet. Påkrevd. Det finnes ingen implisitt standard — serveren gjetter aldri på hjemmekatalogen."
}
},
"required": [
"workspace"
],
"additionalProperties": false
}
}
]
}

114
tests/helpers/mcp_stdio.py Normal file
View file

@ -0,0 +1,114 @@
"""Drive the stdio MCP server in-process, over its own request handler.
The transport MCP specifies is newline-delimited JSON-RPC over stdin and
stdout, but a test that spawned a subprocess to ask `tools/list` would be
measuring the pipe as much as the server, and would need a timeout to avoid
hanging a suite on a handshake that never completes. So this helper speaks to
`handle()` directly: the same dictionaries the loop would have decoded off the
wire, minus the wire. The one thing that buys is that a protocol error is an
assertion failure in the test that caused it, not a hang.
What that deliberately does not cover is the loop itself -- the framing, the
flush, the tolerance for a blank line. `tests/test_mcp_jobbsok_tools.py`
covers those separately, by driving the real launcher as a subprocess and
reading its stdout, which is the only place they are observable.
Style note: this file follows tests/helpers/golden.py.
"""
PROTOCOL_VERSION = "2024-11-05"
class RpcError(AssertionError):
"""A JSON-RPC error response, raised where the test can see the code."""
def __init__(self, method, error):
self.method = method
self.error = error
super().__init__(
"%s returned error %s: %s"
% (method, error.get("code"), error.get("message"))
)
class Client(object):
"""A minimal MCP client bound to one server module.
Ids are assigned here rather than by the caller, because an id the test
chose tells the test nothing: what matters is that the response carries
back the id the request went out with, and that is asserted on every call.
"""
def __init__(self, server):
self.server = server
self._next_id = 0
def request(self, method, params=None):
self._next_id += 1
req_id = self._next_id
message = {"jsonrpc": "2.0", "id": req_id, "method": method}
if params is not None:
message["params"] = params
response = self.server.handle(message)
assert response is not None, "%s got no response; it is not a notification" % method
assert response.get("jsonrpc") == "2.0", "response is not JSON-RPC 2.0: %r" % response
assert response.get("id") == req_id, (
"response id %r does not match request id %r" % (response.get("id"), req_id)
)
if "error" in response:
raise RpcError(method, response["error"])
return response["result"]
def notify(self, method, params=None):
"""Send a notification and assert the server stays silent."""
message = {"jsonrpc": "2.0", "method": method}
if params is not None:
message["params"] = params
assert self.server.handle(message) is None, (
"%s is a notification; the server must not reply to it" % method
)
def initialize(self):
result = self.request(
"initialize",
{
"protocolVersion": PROTOCOL_VERSION,
"capabilities": {},
"clientInfo": {"name": "jobbsok-tests", "version": "0"},
},
)
self.notify("notifications/initialized")
return result
def tools(self):
return self.request("tools/list")["tools"]
def call(self, name, arguments):
"""Call a tool and return the raw result, errors included."""
self._next_id += 1
req_id = self._next_id
response = self.server.handle(
{
"jsonrpc": "2.0",
"id": req_id,
"method": "tools/call",
"params": {"name": name, "arguments": arguments},
}
)
assert response.get("id") == req_id
return response
def call_text(self, name, arguments):
"""Call a tool that is expected to succeed and return its text."""
response = self.call(name, arguments)
if "error" in response:
raise RpcError("tools/call:%s" % name, response["error"])
result = response["result"]
assert result.get("isError") is False, (
"%s reported a tool error: %r" % (name, result)
)
blocks = result["content"]
assert len(blocks) == 1 and blocks[0]["type"] == "text", (
"%s returned %r; one text block was expected" % (name, blocks)
)
return blocks[0]["text"]

View file

@ -0,0 +1,334 @@
"""The `jobbsok-tools` host MCP server and its launcher (plan Step 12).
The server exists so Cowork can reach the same logic the Claude Code CLI
reaches (operator decision 7). That framing decides what is worth testing
here. The arithmetic is already covered by Steps 8 and 10; what is not covered
anywhere else is the seam -- whether the MCP layer hands back what the library
computed, whole and unaltered, or quietly reshapes it on the way out.
Three properties carry that seam, and each has a test of its own:
* **The tool surface is pinned to a golden file.** A tool list is an API. A
renamed argument is a silent break in Cowork, where nothing type-checks the
call, so the list is blessed once and compared byte for byte after.
* **Every tool takes an explicit `workspace`.** `jobbsok_lib.paths` refuses to
guess at a home directory; a server that defaulted the workspace would put
the guess back on the far side of an unsandboxed process (risk C6).
* **`avvisninger` and `advarsler` stay apart, and `vekt_hash` comes along.**
Session 8 decided that a warning is a filter that could not run or a soft
filter that fired, and a rejection is a hard no. Collapsing the two in the
`tools/call` answer would erase divergences 3 and 5 at the transport layer,
where no scoring test would ever see it.
The launcher is tested through a real subprocess in both directions, because
what it exists to prevent -- silently serving on the 3.9.6 that a GUI-spawned
process finds on an empty PATH (risk H2) -- is a property of process startup
and cannot be observed in-process. That test also covers the stdio loop's
framing, which the in-process helper deliberately does not.
Style note: this file follows tests/test_kandidatvurdering_nowrite.py.
"""
import json
import os
import subprocess
import sys
import pytest
import jobbsok_tools
from helpers import mcp_stdio
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
SCRIPTS = os.path.join(REPO, "scripts")
GOLDEN = os.path.join(REPO, "tests", "golden", "jobbsok-tools.tools.json")
LAUNCHER = os.path.join(SCRIPTS, "jobbsok_tools_launch.sh")
PROFIL_FIXTURE = ("profiles", "01-gyldig.md")
ANNONSE_FIXTURE = ("listings", "01-alt-passer.md")
#: A sub-score payload on the 0-100 contract. The corpus registers 0-10 for
#: readability, so the bridge is built here rather than in the corpus, which
#: `test_fixture_hygiene.py` holds to an exact file count.
DELSCORE = {
"delscore": {"fagomrade": 80, "oppgavetype": 70, "teknologi": 60, "selskapstype": 50},
"bekymringer": ["konseptfase, ikke drift"],
}
@pytest.fixture
def klient():
"""An initialized in-process MCP client bound to the server module."""
client = mcp_stdio.Client(jobbsok_tools)
client.initialize()
return client
@pytest.fixture
def arbeidsomrade(empty_workspace, fixtures_dir):
"""A scaffolded workspace holding the valid fixture profile."""
with open(os.path.join(fixtures_dir, *PROFIL_FIXTURE), "r", encoding="utf-8") as handle:
profil = handle.read()
with open(
os.path.join(empty_workspace, "profil", "kandidat.md"), "w", encoding="utf-8"
) as handle:
handle.write(profil)
return empty_workspace
def les_annonse(fixtures_dir):
from jobbsok_lib import frontmatter
with open(os.path.join(fixtures_dir, *ANNONSE_FIXTURE), "r", encoding="utf-8") as handle:
return frontmatter.parse(handle.read())
def test_initialize_and_tools_list_match_the_golden_surface(klient, golden):
hilsen = klient.initialize()
assert hilsen["protocolVersion"] == mcp_stdio.PROTOCOL_VERSION, (
"the server must echo the client's protocol version, not choose its own"
)
assert hilsen["serverInfo"]["name"] == "jobbsok-tools", (
"the server name is what .mcp.json declares; they cannot disagree"
)
assert "tools" in hilsen["capabilities"]
verktoy = klient.tools()
golden(GOLDEN, json.dumps({"tools": verktoy}, ensure_ascii=False, indent=2) + "\n")
# The golden pins the shape; this pins the fact that M1 ships exactly three
# tools, so Step 24 adding four more is a visible change and not a drift.
assert [t["name"] for t in verktoy] == ["kandidat_valider", "annonse_vurder", "selvsjekk"]
def test_every_tool_requires_an_explicit_workspace(klient, arbeidsomrade):
for verktoy in klient.tools():
skjema = verktoy["inputSchema"]
assert "workspace" in skjema["properties"], (
"%s does not take a workspace argument" % verktoy["name"]
)
assert "workspace" in skjema.get("required", []), (
"%s does not require workspace; there is no implicit default"
% verktoy["name"]
)
# Declared as required is one thing; refused at the call is the thing
# that matters, because Cowork does not enforce the schema for us.
svar = klient.call(verktoy["name"], {})
assert "error" in svar or svar["result"].get("isError") is True, (
"%s answered a call with no workspace instead of refusing it"
% verktoy["name"]
)
# And a workspace that escapes its root is refused by the same check the
# library uses, rather than being read from wherever it resolved.
svar = klient.call(
"kandidat_valider", {"workspace": arbeidsomrade, "sti": "../../etc/passwd"}
)
assert svar["result"]["isError"] is True
assert "workspace" in svar["result"]["content"][0]["text"].lower()
def test_the_server_and_the_library_produce_identical_output(
klient, arbeidsomrade, fixtures_dir
):
import kandidat_schema
import vurdering
fra_server = klient.call_text(
"kandidat_valider", {"workspace": arbeidsomrade, "sti": "profil/kandidat.md"}
)
fra_bibliotek = kandidat_schema.report_json(
kandidat_schema.validate_file(arbeidsomrade, "profil", "kandidat.md")
)
assert fra_server == fra_bibliotek, (
"the MCP layer reshaped the validation report on its way out"
)
annonse, brodtekst = les_annonse(fixtures_dir)
fra_server = klient.call_text(
"annonse_vurder",
{
"workspace": arbeidsomrade,
"annonse": annonse,
"brodtekst": brodtekst,
"delscore": DELSCORE["delscore"],
"bekymringer": DELSCORE["bekymringer"],
},
)
profil = vurdering.les_profil_fil(arbeidsomrade, "profil", "kandidat.md")
fra_bibliotek = jobbsok_tools.til_json(
vurdering.vurder(profil, annonse, brodtekst, DELSCORE)
)
assert fra_server == fra_bibliotek, (
"the MCP layer reshaped the scoring result on its way out"
)
def test_the_scoring_tool_keeps_avvisninger_advarsler_and_vekt_hash_apart(
klient, arbeidsomrade, fixtures_dir
):
annonse, brodtekst = les_annonse(fixtures_dir)
# A listing that passes every hard filter but leaves salary unstated: one
# warning, no rejection. Collapsing the two would make this read as a no.
uten_lonn = dict(annonse)
uten_lonn.pop("lonn_nok")
resultat = json.loads(
klient.call_text(
"annonse_vurder",
{
"workspace": arbeidsomrade,
"annonse": uten_lonn,
"brodtekst": brodtekst,
"delscore": DELSCORE["delscore"],
},
)
)
assert resultat["verdikt"] == "vurderes"
assert resultat["avvisninger"] == []
assert any("lonn" in a["nokkel"] for a in resultat["advarsler"]), (
"an unstated salary must surface as a warning, not vanish: %r"
% (resultat["advarsler"],)
)
assert resultat["vekt_hash"].startswith("sha256:")
import vurdering
assert resultat["vekt_hash"] == vurdering.vekt_hash(resultat["vekter"]), (
"vekt_hash must be the fingerprint of the weights actually used"
)
def test_importing_the_server_does_not_import_the_capability_modules():
"""Cold start is a budget: Cowork times an MCP server out on startup."""
proc = subprocess.run(
[
sys.executable,
"-c",
"import sys; import jobbsok_tools; "
"print(sorted(m for m in ('kandidat_schema', 'vurdering', "
"'llm_ingestion_guard') if m in sys.modules))",
],
cwd=REPO,
env=dict(os.environ, PYTHONPATH=SCRIPTS),
capture_output=True,
text=True,
)
assert proc.returncode == 0, proc.stderr
assert proc.stdout.strip() == "[]", (
"importing the server pulled in %s at module load; the capability "
"modules must be imported inside the handlers" % proc.stdout.strip()
)
def test_selvsjekk_reports_the_interpreter_and_the_pinned_guard(klient, arbeidsomrade):
rapport = json.loads(klient.call_text("selvsjekk", {"workspace": arbeidsomrade}))
assert rapport["python_executable"] == sys.executable
major, minor = rapport["python_version_info"][:2]
assert (major, minor) >= (3, 10), (
"the server is serving on Python %r; 3.10+ is the floor"
% (rapport["python_version"],)
)
assert rapport["guard_versjon"] == "1.3.0", (
"selvsjekk must report the guard version actually installed, and the "
"pin in pyproject.toml is v1.3.0; got %r" % (rapport["guard_versjon"],)
)
assert rapport["workspace"] == os.path.realpath(arbeidsomrade)
# BUILD_STAMP arrives in Step 13. Whether it is there or not, saying so
# plainly is the contract -- a missing stamp must never read as a match.
assert "build_stamp" in rapport
def test_the_launcher_refuses_an_interpreter_below_3_10_and_serves_on_a_good_one(tmp_path):
falsk = tmp_path / "python3.9"
falsk.write_text(
"#!/bin/bash\n"
"# Answers the version probe as 3.9.6 and fails the version gate.\n"
'case "$*" in\n'
" *print*) echo '3.9.6'; exit 0 ;;\n"
"esac\n"
"exit 1\n"
)
falsk.chmod(0o755)
avvist = subprocess.run(
["bash", LAUNCHER],
env=dict(os.environ, JOBBSOK_PYTHON=str(falsk)),
input="",
capture_output=True,
text=True,
)
assert avvist.returncode != 0, (
"the launcher started on a 3.9 interpreter instead of refusing"
)
assert "3.10" in avvist.stderr, (
"the refusal must name the floor it enforced: %r" % (avvist.stderr,)
)
# With no candidate at all, the launcher must refuse rather than take
# whatever python3 the PATH offers -- which under an empty launchd PATH on
# this Mac is 3.9.6 (risk H2). Nothing in the branch above reaches this
# path, so it needs its own case: a mutation that reinstated an ungated
# PATH fallback survived the JOBBSOK_PYTHON case untouched.
falsk_bin = tmp_path / "bin"
falsk_bin.mkdir()
(falsk_bin / "python3").write_text(falsk.read_text())
(falsk_bin / "python3").chmod(0o755)
tom_rot = tmp_path / "tom-plugin-rot"
(tom_rot / "scripts").mkdir(parents=True)
miljo = dict(os.environ)
miljo.pop("JOBBSOK_PYTHON", None)
miljo.pop("CLAUDE_PLUGIN_DATA", None)
miljo["CLAUDE_PLUGIN_ROOT"] = str(tom_rot)
miljo["PATH"] = "%s:/usr/bin:/bin" % falsk_bin
uten_kandidat = subprocess.run(
["/bin/bash", LAUNCHER],
env=miljo,
input="",
capture_output=True,
text=True,
)
assert uten_kandidat.returncode != 0, (
"the launcher fell through to the python3 on PATH instead of refusing"
)
assert "refusing to start" in uten_kandidat.stderr, (
"the refusal must say so plainly: %r" % (uten_kandidat.stderr,)
)
# The other direction, over the real stdio loop: framing, flush and all.
forespoersler = (
json.dumps(
{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {"protocolVersion": mcp_stdio.PROTOCOL_VERSION},
}
)
+ "\n"
+ json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"})
+ "\n"
+ "\n" # a blank line the loop must tolerate rather than answer
+ json.dumps({"jsonrpc": "2.0", "id": 2, "method": "tools/list"})
+ "\n"
)
servert = subprocess.run(
["bash", LAUNCHER],
env=dict(os.environ, JOBBSOK_PYTHON=sys.executable),
input=forespoersler,
capture_output=True,
text=True,
timeout=30,
)
assert servert.returncode == 0, servert.stderr
svar = [json.loads(line) for line in servert.stdout.splitlines() if line.strip()]
assert [s["id"] for s in svar] == [1, 2], (
"expected exactly two replies, one per request: %r" % (servert.stdout,)
)
assert svar[0]["result"]["serverInfo"]["name"] == "jobbsok-tools"
assert [t["name"] for t in svar[1]["result"]["tools"]] == [
"kandidat_valider",
"annonse_vurder",
"selvsjekk",
]