#!/usr/bin/env python3 """ebf_mcp_server.py — DW-1352. The Model Context Protocol server for eBilanz Fabrik. WHAT THIS IS FOR ---------------- David, 2026-09-24: *"what about developing an MCP for other AI agents to easily connect with eBilanz Fabrik and file e-Bilanzen"* — on the day a customer arrived from ChatGPT and filed, and with chatgpt.com level with all of Google in our attributed paid book (8 against 9, first 2026-07-30). Content gets us cited; an MCP puts us INSIDE the assistant as a connector its user installs. That is distribution rather than visibility. WHAT IT DELIBERATELY DOES NOT DO — AND WHY THAT IS NOT A LIMITATION ------------------------------------------------------------------- IT NEVER FILES. It never accepts an ELSTER certificate or a PIN, never transmits to ELSTER, never takes a payment, and offers no unattended filing path. That is a design decision and it is a commitment we have already PUBLISHED: `llms.txt` tells every engine that reads it "no unattended filing API is offered", and llms-full.txt states the reason — deciding which position goes where is, when done as a business for someone else, reserved to the tax-advising professions (§§ 2-5 StBerG); eBilanz Fabrik does not make that decision case by case. For some positions the tool asks the customer; totals it does not ask about are filed under a standard position by one fixed rule that is the same for everyone; the customer releases the filing. What is left is almost all of the work. TYPING THE FIGURES is the job, and an agent may do it: the agent checks eligibility, validates the numbers against the same engine the funnel uses, and hands its person a link with everything already filled in. The person reviews, pays, and signs with their own certificate. THE LINK IS A FRAGMENT, DELIBERATELY. Browsers never transmit `#…`, so the figures an agent puts in a link never reach a request log, a referrer header or an access log — the same property `ebf-prefill-link.py` was built around (DW-428). An agent handing its person a link is not sending us their tax figures: the link carries no figures to us. The check does — /api/ampel-check drops them on return and stores only an anonymised pattern. ONE SEAM, ONE ANSWER. Eligibility and the figure checks are not re-implemented here: they are asked of `/api/ampel-check`, the public endpoint that reuses the production engine. Prices are read from the `Offer` JSON-LD on `/guide/fuer-ki-assistenten.html`, which is generated from the same `_OFFER` table the funnel prices from. A second implementation of either would drift from the thing it describes, and this file would become another place that has to be remembered. NO DEPENDENCIES. Plain JSON-RPC 2.0 over stdio, stock Python 3 — so any agent runtime can launch it without a package install, which is the whole point of "easily connect". Usage in a client config: {"command": "python3", "args": ["/path/to/ebf_mcp_server.py"]} Self-check: python3 ebf_mcp_server.py --self-test """ from __future__ import annotations import base64 import json import re import sys import urllib.error import urllib.request APEX = "https://ebilanzfabrik.de" APP = "https://app.ebilanzfabrik.de" AGENT_PAGE = APEX + "/guide/fuer-ki-assistenten.html" UA = "ebf-mcp-server/1.0 (+https://ebilanzfabrik.de/guide/fuer-ki-assistenten.html)" TIMEOUT = 25 BOUNDARY = ("This server never files, never takes payment, and never accepts an ELSTER certificate " "or PIN. It prepares; the human reviews, pays and signs.") SIZE_CLASS_NOTE = ("The size class is stated by the customer and is their responsibility. eBilanz Fabrik cannot " "verify it (AGB § 2a (2)).") PROTOCOL_VERSIONS = ("2025-11-25", "2025-06-18", "2025-03-26", "2024-11-05") _READ_ONLY = {"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False} _RECORDS_A_PATTERN = {"readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": False} def _get(url: str) -> bytes: req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "application/json,*/*"}) with urllib.request.urlopen(req, timeout=TIMEOUT) as r: return r.read() def _post(url: str, payload: dict) -> dict: body = json.dumps(payload, ensure_ascii=False).encode("utf-8") req = urllib.request.Request(url, data=body, method="POST", headers={"User-Agent": UA, "Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=TIMEOUT) as r: return json.loads(r.read().decode("utf-8")) except urllib.error.HTTPError as e: try: detail = json.loads(e.read().decode("utf-8")) except Exception: detail = None return {"_refused": True, "_http": int(e.code), "_detail": detail} _LEGAL_FORMS = ("GmbH", "UG", "UG (haftungsbeschränkt)", "AG", "Aktiengesellschaft", "SE", "Societas Europaea", "andere") _SIZE_CLASSES = ("Kleinst", "klein", "mittelgross", "gross") def _canon(value, allowed): """Case-fold an agent's token onto the engine's spelling — and pass anything else through. DW-1364. Measured against production: `groesse="Kleinst"` answers `eligible: true`, and `groesse="kleinst"` answers `eligible: false` with the reason *"Größenklasse 'kleinst' is out of MVP scope — eBilanz Fabrik covers kleine + Kleinst Kapitalgesellschaft only"* — a refusal that contradicts itself in its own sentence, on the money path, for a company we serve. Same for `gmbh` against `GmbH`. An agent lower-cases enum values as a matter of course. An UNKNOWN token is returned untouched on purpose: the engine stays the authority on who can be served, and this must never invent an eligibility by snapping a typo onto a real legal form. """ v = str(value or "").strip() if not v: return value for a in allowed: if v.casefold() == a.casefold(): return a return value def _declined(res: dict, tool: str, sent_figures: bool) -> dict | None: """A 200 that carries NO JUDGEMENT. DW-1368. `_refusal` below catches an HTTP error. It does not catch the refusals the engine answers **200** with, and there are three: `ampel_check`'s defensive `except Exception:` seam, the route's `_human` fallback (>400 keys, or any engine exception), and the `_disabled` kill-switch. All three return `checks: {}`, `reasons: []`, `balance.aktiva: null` — and `t_validate` rendered them as an answer with `checks_failed: []` beside the sentence *"Fix what is in checks_failed"*, which is empty. Measured live before this existed: a Bilanz of `{kasse 1.000,00, gezeichnetes_kapital 1.000,00}` and the same one with `gezeichnetes_kapital 500,00` — off by 100 % — produced **byte-identical** tool output. That is verbatim the symptom DW-1364 was written to kill, one layer down: the 422 route was closed and the 200 routes were never driven. A rejection is a routing decision, and this is where it routed. THE DISCRIMINATOR, verified against production with both controls: * judged -> `checks` carries its four keys AND `balance.aktiva` is parsed (`pass` on a balanced fixture, `issue` + `bilanzgleichung: false` on one 10 000 off) * legibly out -> `checks` empty BUT `reasons` non-empty (`eligible: false`) * NOT an answer -> `checks` empty AND `reasons` empty, while figures were sent Only the third is caught here, so a real verdict and a real refusal both pass through untouched. """ if not isinstance(res, dict) or not sent_figures: return None if (res.get("checks") or {}) or (res.get("reasons") or []): return None return {"engine_declined_to_judge": True, "raw_verdict": res.get("verdict"), "what_this_means": ( "The engine answered, but it did NOT judge these figures: it returned no checks and " "no reasons, and read no balance. Nothing here is a pass. `%s` withholds " "`checks_failed` in this case on purpose — an empty failure list beside a verdict " "reads as 'all clear', and for these inputs it would be a balance sheet nobody " "looked at. Send the full §266/§275 line items (summe_aktiva, summe_passiva and the " "equity positions) and ask again." % tool), "boundary": BOUNDARY} _JUDGED_VERDICTS = ("pass", "issue", "review", "outofscope") def _not_judged(res: dict, tool: str) -> dict | None: """A 200 whose verdict is not a judgement (DW-1784). None when it is one.""" if not isinstance(res, dict) or res.get("verdict") in _JUDGED_VERDICTS: return None return {"engine_declined_to_judge": True, "raw_verdict": res.get("verdict"), "what_this_means": ( "The eligibility check did not judge this company: the engine answered verdict %r, " "which is not a judgement (`disabled` means the check is switched off on the server). " "`%s` returns no `eligible` in this case on purpose — the engine's placeholder says " "true, and that is not an answer. Nothing here says this company can be served; ask " "again later." % (res.get("verdict"), tool)), "boundary": BOUNDARY} def _refusal(res: dict, tool: str) -> dict | None: """Render an engine refusal so that no field of it can be read as 'nothing was wrong'.""" if not isinstance(res, dict) or not res.get("_refused"): return None return {"engine_refused": True, "http_status": res.get("_http"), "detail": res.get("_detail"), "what_this_means": ("The engine REFUSED this request — it did not examine the figures, " "so nothing here is a pass. `%s` returns no verdict in this case by " "design: a null verdict beside an empty `checks_failed` would read " "as 'all clear'. Correct the request and ask again." % tool), "boundary": BOUNDARY} def t_products(_args: dict) -> dict: """Which products the server will actually sell right now — asked of the server, not asserted.""" return json.loads(_get(APP + "/api/products").decode("utf-8")) def t_pricing(_args: dict) -> dict: """Prices, read from the Offer JSON-LD the site publishes for exactly this purpose. Not a constant in this file: that would be a second copy of a number customers are charged, and it would go stale silently the first time a price moved. """ html = _get(AGENT_PAGE).decode("utf-8", "replace") offers = [] for blk in re.findall(r'', html, re.S): try: doc = json.loads(blk) except Exception: continue for node in (doc.get("@graph") or [doc]): for off in (node.get("offers") or []): offers.append({"name": off.get("name"), "price": off.get("price"), "currency": off.get("priceCurrency"), "vat_included": off.get("valueAddedTaxIncluded"), "url": off.get("url")}) if not offers: raise RuntimeError("no Offer found on %s — refusing to quote a price I could not read " "(a number in customer prose needs a source you can point at)" % AGENT_PAGE) return {"offers": offers, "source": AGENT_PAGE, "note": "Net prices, plus 19 % German VAT, one-off per submission, no subscription."} _ELIG_MAP = { "rechtsform": "legal_form", "legal_form": "legal_form", "groesse": "size_class", "size_class": "size_class", "wirtschaftsjahr": "fy", "fy": "fy", "stichtag": "stichtag", "rechenwerk": "rechenwerk", "prior_year_equity": "prior_year_equity", "bilanz": "bilanz", "guv": "guv", } _INT_FIELDS = ("fy",) def _to_engine(args: dict) -> dict: out, unknown = {}, [] for k, v in (args or {}).items(): if v in (None, ""): continue tgt = _ELIG_MAP.get(k) if not tgt: unknown.append(k) continue out[tgt] = int(str(v)) if tgt in _INT_FIELDS else v if unknown: raise ValueError("unknown field(s) %s — this server refuses to forward a name the engine " "would silently drop. Known: %s" % (sorted(unknown), sorted(set(_ELIG_MAP)))) return out def _prepare(args: dict) -> dict: """Map an agent's argument object onto the engine's, with the two DW-1364 corrections.""" payload = _to_engine(args) if "legal_form" in payload: payload["legal_form"] = _canon(payload["legal_form"], _LEGAL_FORMS) if "size_class" in payload: payload["size_class"] = _canon(payload["size_class"], _SIZE_CLASSES) for _f, _de in (("legal_form", "rechtsform"), ("size_class", "groesse")): if not str(payload.get(_f) or "").strip(): raise ValueError( "%s is required — the engine DEFAULTS it (legal_form=GmbH, size_class=Kleinst), so " "leaving it out does not ask a question, it answers a different one. Give the " "company's own values: rechtsform one of %s, groesse one of %s." % (_de, ", ".join(_LEGAL_FORMS), ", ".join(_SIZE_CLASSES))) if not str(payload.get("stichtag") or "").strip(): raise ValueError("stichtag is required — give the balance sheet date as YYYY-MM-DD " "(e.g. 2025-12-31). Without it the engine reports the Wirtschaftsjahr as " "abweichend and answers eligible=false for a company it would otherwise " "serve.") return payload def t_eligibility(args: dict) -> dict: """Ask the production engine whether this company can be served, and why not if not.""" payload = _prepare(args) res = _post(APP + "/api/ampel-check", payload) ref = _refusal(res, "ebf_check_eligibility") or _not_judged(res, "ebf_check_eligibility") if ref: ref["asked"] = payload return ref return {"verdict": res.get("verdict"), "eligible": res.get("eligible"), "reasons": res.get("reasons") or [], "checks": res.get("checks") or {}, "asked": payload, "size_class": SIZE_CLASS_NOTE, "boundary": BOUNDARY} def t_validate(args: dict) -> dict: """Run the customer's figures past the SAME checks the funnel runs, before anyone starts. This is the tool that earns an agent its keep: the balance, the GuV tie-out and the eligibility rules are the things a person gets wrong and then discovers at the payment step. """ payload = _prepare(args) res = _post(APP + "/api/ampel-check", payload) ref = _refusal(res, "ebf_validate_figures") or _declined( res, "ebf_validate_figures", bool(payload.get("bilanz") or payload.get("guv"))) if ref: ref["asked"] = {k: v for k, v in payload.items() if k not in ("bilanz", "guv")} return ref checks = res.get("checks") or {} return {"verdict": res.get("verdict"), "eligible": res.get("eligible"), "needs_review": res.get("needs_review"), "size_band": res.get("size_band"), "reasons": res.get("reasons") or [], "balance": res.get("balance") or {}, "checks": checks, "checks_failed": sorted(k for k, v in checks.items() if v is False), "checks_not_evaluated": sorted(k for k, v in checks.items() if v is None), "fails": res.get("fails") or [], "how_to_read_this": "checks_not_evaluated were NOT taken — the engine stops at the first " "structural failure. A null is not a pass. Fix what is in " "checks_failed, then ask again to learn about the rest.", "boundary": BOUNDARY} def t_link(args: dict) -> dict: """Build the link the agent hands its person: every field already filled. The payload rides in a URL FRAGMENT, which browsers never transmit — so the figures do not reach our logs, a referrer header, or anyone else, until the person opens the page themselves. """ produkt = str(args.get("produkt") or "").strip() if produkt and produkt not in ("ebilanz", "offenlegung", "paket"): raise ValueError("produkt must be one of ebilanz, offenlegung, paket — got %r" % produkt) fields = args.get("fields") or {} if not isinstance(fields, dict) or not fields: raise ValueError("fields must be a non-empty object of funnel field ids to values") payload = {"fields": fields} if args.get("elig"): payload["elig"] = args["elig"] if args.get("step"): payload["step"] = args["step"] raw = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") tok = base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") q = ("?produkt=%s" % produkt) if produkt else "" return { "url": "%s/start.html%s#p=%s" % (APEX, q, tok), "fields_carried": len(fields), "what_the_human_still_does": [ "review every figure on screen — nothing is filed from this link", "pay (Stripe, one-off, no subscription)", "release the transmission with their own ELSTER certificate and PIN", ], "privacy": "The values ride in the URL fragment. Browsers never send '#…' to a server, so " "they stay on the person's machine until they open the page themselves.", "boundary": BOUNDARY, } TOOLS = [ {"name": "ebf_get_products", "fn": t_products, "title": "Products available now", "annotations": dict(_READ_ONLY, title="Products available now"), "description": "Which eBilanz Fabrik products can be bought right now, as the server itself " "reports. " + BOUNDARY, "inputSchema": {"type": "object", "properties": {}}}, {"name": "ebf_get_pricing", "fn": t_pricing, "title": "Current prices", "annotations": dict(_READ_ONLY, title="Current prices"), "description": "Current net prices per submission, read from the site's published Offer " "metadata. German VAT of 19 % is added at checkout. " + BOUNDARY, "inputSchema": {"type": "object", "properties": {}}}, {"name": "ebf_check_eligibility", "fn": t_eligibility, "title": "Check eligibility", "annotations": dict(_RECORDS_A_PATTERN, title="Check eligibility"), "description": "Ask whether a given company can be served — legal form, size class, fiscal " "year — answered by the production engine, with the reasons if not. " + BOUNDARY, "inputSchema": {"type": "object", "required": ["rechtsform", "groesse", "stichtag"], "properties": { "rechtsform": {"type": "string", "enum": list(_LEGAL_FORMS), "description": "legal form — " + ", ".join(_LEGAL_FORMS) + " (case does not matter)"}, "groesse": {"type": "string", "enum": list(_SIZE_CLASSES), "description": "size class per § 267/267a HGB — " + ", ".join(_SIZE_CLASSES) + " (case does not matter)"}, "wirtschaftsjahr": {"type": "string", "description": "the fiscal year, e.g. 2024"}, "stichtag": {"type": "string", "description": "REQUIRED — balance sheet date as YYYY-MM-DD, e.g. 2024-12-31. " "Omitting it makes the engine treat the year as abweichend " "and answer eligible=false."}}}}, {"name": "ebf_validate_figures", "fn": t_validate, "title": "Validate figures", "annotations": dict(_RECORDS_A_PATTERN, title="Validate figures"), "description": "Check a balance sheet and P&L against the same rules the funnel applies — " "balance, P&L tie-out, eligibility — BEFORE the person starts. Catches what " "people otherwise discover at the payment step. " + BOUNDARY, "inputSchema": {"type": "object", "required": ["legal_form", "size_class", "stichtag"], "properties": { "bilanz": {"type": "object", "additionalProperties": {"type": "string"}, "description": "§ 266 HGB line-item keys to values as STRINGS in German number " "notation, e.g. {\"umsatzerloese\": \"1.234,56\"}. A numeric " "value is refused by the engine."}, "guv": {"type": "object", "additionalProperties": {"type": "string"}, "description": "§ 275 HGB line-item keys to values as STRINGS, same notation"}, "legal_form": {"type": "string", "enum": list(_LEGAL_FORMS)}, "size_class": {"type": "string", "enum": list(_SIZE_CLASSES)}, "fy": {"type": "integer"}, "stichtag": {"type": "string", "description": "REQUIRED — YYYY-MM-DD"}, "rechenwerk": {"type": "string", "description": "steuerbilanz (default) or handelsbilanz"}}}}, {"name": "ebf_build_filing_link", "fn": t_link, "title": "Build a pre-filled link", "annotations": dict(_READ_ONLY, title="Build a pre-filled link"), "description": "Build a link that opens the funnel with every field already filled in, to hand " "to the person. The figures ride in the URL fragment and never reach a server " "log. This does NOT file anything: the person reviews, pays, and releases the " "transmission with their own ELSTER certificate. " + BOUNDARY, "inputSchema": {"type": "object", "properties": { "fields": {"type": "object", "description": "funnel field ids to values"}, "elig": {"type": "object"}, "produkt": {"type": "string", "description": "ebilanz | offenlegung | paket"}, "step": {"type": "integer"}}, "required": ["fields"]}}, ] BY_NAME = {t["name"]: t for t in TOOLS} REMOTE_TOOL_NAMES = ("ebf_get_products", "ebf_get_pricing", "ebf_check_eligibility") _LISTED_KEYS = ("name", "title", "description", "inputSchema", "annotations") def _annotation_problems(tools) -> list: """Every listed tool states a title and the four hints, and none claims to be destructive. Used by both self-tests (this file and the remote endpoint's) — a list nobody checks is prose.""" out = [] for t in tools: a = t.get("annotations") or {} if not t.get("title"): out.append("%s has no title" % t.get("name")) missing = [k for k in ("readOnlyHint", "destructiveHint", "idempotentHint", "openWorldHint") if k not in a] if missing: out.append("%s lacks annotation(s) %s" % (t.get("name"), missing)) if a.get("destructiveHint") is not False: out.append("%s does not state destructiveHint: false" % t.get("name")) return out def _result(rid, payload): return {"jsonrpc": "2.0", "id": rid, "result": payload} def handle(msg: dict, tools: list | None = None) -> dict | None: """One JSON-RPC message in, one reply (or None) out. `tools` is the set to expose — the stdio server passes nothing and gets all five; the remote endpoint passes its own public-data set (DW-1753), so a tool it does not list cannot be called either: it answers -32601, like a tool that does not exist.""" tools = TOOLS if tools is None else tools by_name = {t["name"]: t for t in tools} m, rid = msg.get("method"), msg.get("id") if "id" not in msg: return None if m == "ping": return _result(rid, {}) if m == "initialize": want = str(((msg.get("params") or {}).get("protocolVersion")) or "") return _result(rid, {"protocolVersion": want if want in PROTOCOL_VERSIONS else PROTOCOL_VERSIONS[0], "capabilities": {"tools": {"listChanged": False}}, "serverInfo": {"name": "ebilanz-fabrik", "title": "eBilanz Fabrik", "version": "1.1.0"}, "instructions": BOUNDARY}) if m in ("notifications/initialized", "initialized"): return None if m == "tools/list": return _result(rid, {"tools": [{k: t[k] for k in _LISTED_KEYS if k in t} for t in tools]}) if m == "tools/call": p = msg.get("params") or {} tool = by_name.get(p.get("name")) if not tool: return {"jsonrpc": "2.0", "id": rid, "error": {"code": -32601, "message": "unknown tool %r" % p.get("name")}} try: out = tool["fn"](p.get("arguments") or {}) return _result(rid, {"content": [{"type": "text", "text": json.dumps(out, ensure_ascii=False, indent=1)}]}) except Exception as exc: return _result(rid, {"isError": True, "content": [{"type": "text", "text": "%s: %s" % (type(exc).__name__, exc)}]}) return {"jsonrpc": "2.0", "id": rid, "error": {"code": -32601, "message": "unknown method %r" % m}} def serve() -> int: for line in sys.stdin: line = line.strip() if not line: continue try: msg = json.loads(line) except Exception: continue out = handle(msg) if out is not None: sys.stdout.write(json.dumps(out, ensure_ascii=False) + "\n") sys.stdout.flush() return 0 def self_test() -> int: fails = [] init = handle({"jsonrpc": "2.0", "id": 1, "method": "initialize"}) if not (init and init["result"]["serverInfo"]["name"] == "ebilanz-fabrik"): fails.append("initialize did not identify the server") if handle({"jsonrpc": "2.0", "method": "notifications/initialized"}) is not None: fails.append("a notification must not be answered") lst = handle({"jsonrpc": "2.0", "id": 2, "method": "tools/list"}) names = [t["name"] for t in lst["result"]["tools"]] if len(names) != 5: fails.append("expected 5 tools, got %d: %s" % (len(names), names)) unk = handle({"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "ebf_file_it_for_me", "arguments": {}}}) if "error" not in unk: fails.append("an unknown tool must be an error, not a silent empty result") for t in TOOLS: if "never files" not in t["description"]: fails.append("%s: its description does not state the filing boundary — an agent reads " "descriptions, so that is where the boundary has to live" % t["name"]) banned = ("zertifikat", "certificate", "pin", "passwort", "password") for t in TOOLS: props = ((t["inputSchema"] or {}).get("properties") or {}) for k in props: if any(b in k.lower() for b in banned): fails.append("%s accepts %r — this server must never take a credential" % (t["name"], k)) ok = t_link({"fields": {"f_name": "Musterbetrieb GmbH", "kasse": "5.774,25"}, "produkt": "ebilanz", "step": 3}) if "#p=" not in ok["url"] or "?produkt=ebilanz" not in ok["url"]: fails.append("the link lost its fragment or its product: %s" % ok["url"]) tok = ok["url"].split("#p=", 1)[1] back = json.loads(base64.urlsafe_b64decode(tok + "=" * (-len(tok) % 4))) if back.get("fields", {}).get("kasse") != "5.774,25": fails.append("the payload does not round-trip: %r" % back) if "?" in ok["url"].split("#", 1)[1]: fails.append("query material leaked into the fragment") for bad, why in (({}, "empty fields"), ({"fields": {}}, "empty fields dict"), ({"fields": {"a": 1}, "produkt": "nope"}, "unknown product")): try: t_link(bad) fails.append("accepted %s — it must refuse" % why) except (ValueError, AttributeError): pass mapped = _to_engine({"rechtsform": "GmbH", "groesse": "Kleinst", "wirtschaftsjahr": "2024"}) if mapped != {"legal_form": "GmbH", "size_class": "Kleinst", "fy": 2024}: fails.append("the friendly names do not map onto the engine's: %r" % mapped) if not isinstance(mapped.get("fy"), int): fails.append("fy must reach the engine as an int, got %r" % type(mapped.get("fy"))) try: _to_engine({"wirtschaftjahr": "2024"}) fails.append("a misspelled field was forwarded — pydantic would drop it silently") except ValueError: pass for got, want, why in ((_canon("gmbh", _LEGAL_FORMS), "GmbH", "lower-case legal form"), (_canon("KLEINST", _SIZE_CLASSES), "Kleinst", "upper-case size class"), (_canon("Limited", _LEGAL_FORMS), "Limited", "UNKNOWN form must pass through untouched"), (_canon("", _LEGAL_FORMS), "", "empty stays empty")): if got != want: fails.append("_canon %s: got %r, want %r" % (why, got, want)) ref = _refusal({"_refused": True, "_http": 422, "_detail": {"detail": [{"msg": "x"}]}}, "t") if not (ref and ref.get("engine_refused") is True and ref.get("http_status") == 422): fails.append("a 422 did not render as a refusal: %r" % ref) if _refusal({"verdict": "review", "checks": {}}, "t") is not None: fails.append("a genuine answer was mistaken for a refusal") import io as _io _real_urlopen = urllib.request.urlopen try: def _raise_422(*_a, **_k): raise urllib.error.HTTPError("http://x", 422, "Unprocessable", {}, _io.BytesIO(b'{"detail":[{"msg":"Input should be a valid string"}]}')) urllib.request.urlopen = _raise_422 _r = _post("http://x", {}) if not (_r.get("_refused") is True and _r.get("_http") == 422 and _r.get("_detail")): fails.append("_post did not label a 422 as a refusal: %r" % _r) def _raise_500(*_a, **_k): raise urllib.error.HTTPError("http://x", 500, "Server Error", {}, _io.BytesIO(b"not json")) urllib.request.urlopen = _raise_500 _r5 = _post("http://x", {}) if not (_r5.get("_refused") is True and _r5.get("_http") == 500): fails.append("_post did not label an unparseable 500 as a refusal: %r" % _r5) finally: urllib.request.urlopen = _real_urlopen _real_post = globals()["_post"] try: globals()["_post"] = lambda *_a, **_k: {"_refused": True, "_http": 422, "_detail": {"d": 1}} out = t_validate({"bilanz": {"umsatzerloese": "1,00"}, "stichtag": "2025-12-31", "fy": 2025, "legal_form": "GmbH", "size_class": "Kleinst"}) if not out.get("engine_refused"): fails.append("a refused validate still rendered as an answer: %r" % sorted(out)) if "checks_failed" in out or out.get("verdict", "sentinel") is None: fails.append("a refused validate still offers the fields that read as 'all clear'") finally: globals()["_post"] = _real_post for note in ({"jsonrpc": "2.0", "method": "notifications/cancelled", "params": {"requestId": 7}}, {"jsonrpc": "2.0", "method": "notifications/progress"}, {"jsonrpc": "2.0", "method": "entirely/unknown"}): if handle(note) is not None: fails.append("a notification (%s) got a reply — JSON-RPC 2.0 forbids it" % note["method"]) pong = handle({"jsonrpc": "2.0", "id": 41, "method": "ping"}) if not (pong and pong.get("id") == 41 and pong.get("result") == {}): fails.append("ping was not answered with an empty result: %r" % pong) bad = handle({"jsonrpc": "2.0", "id": 42, "method": "entirely/unknown"}) if not (bad and (bad.get("error") or {}).get("code") == -32601 and bad.get("id") == 42): fails.append("a REQUEST for an unknown method lost its -32601 — the notification rule swallowed it") JUDGED = {"verdict": "issue", "checks": {"bilanzgleichung": False}, "reasons": [], "balance": {"aktiva": "100000.00"}} OUTOFSCOPE = {"verdict": "outofscope", "eligible": False, "checks": {}, "reasons": ["Rechtsform 'Limited' is out of MVP scope"]} NONANSWER = {"verdict": "issue", "eligible": True, "checks": {}, "reasons": [], "balance": {"aktiva": None, "passiva": None, "balanced": False}} if _declined(NONANSWER, "t", True) is None: fails.append("a 200 carrying no checks and no reasons was taken for an answer") if _declined(JUDGED, "t", True) is not None: fails.append("a REAL verdict was mistaken for a non-answer") if _declined(OUTOFSCOPE, "t", True) is not None: fails.append("a legible out-of-scope refusal was mistaken for a non-answer") if _declined(NONANSWER, "t", False) is not None: fails.append("an eligibility call carrying no figures was flagged — it legitimately has neither") _real_post2 = globals()["_post"] try: globals()["_post"] = lambda *_a, **_k: dict(NONANSWER) out2 = t_validate({"bilanz": {"kasse": "1,00"}, "stichtag": "2025-12-31", "fy": 2025, "legal_form": "GmbH", "size_class": "Kleinst"}) if not out2.get("engine_declined_to_judge"): fails.append("t_validate rendered a non-answer as an answer: %r" % sorted(out2)) if "checks_failed" in out2: fails.append("a non-answer still offers checks_failed, which reads as 'all clear'") finally: globals()["_post"] = _real_post2 DISABLED = {"verdict": "disabled", "eligible": True, "checks": {}, "fails": [], "reasons": [], "needs_review": False, "size_band": "unknown"} INSCOPE = {"verdict": "review", "eligible": True, "reasons": [], "needs_review": True, "checks": {"bilanzgleichung": None}} _real_post3 = globals()["_post"] try: for shape, want_answer in ((DISABLED, False), (INSCOPE, True), (OUTOFSCOPE, True)): globals()["_post"] = lambda *_a, _s=shape, **_k: dict(_s) out3 = t_eligibility({"legal_form": "GmbH", "size_class": "Kleinst", "fy": 2025, "stichtag": "2025-12-31"}) if want_answer and "eligible" not in out3: fails.append("a real eligibility answer (%s) was withheld" % shape["verdict"]) if want_answer and out3.get("size_class") != SIZE_CLASS_NOTE: fails.append("an eligibility answer (%s) does not name the size class as the customer's own " "statement (Hugo D4)" % shape["verdict"]) if not want_answer and ("eligible" in out3 or not out3.get("engine_declined_to_judge")): fails.append("the kill-switch answer still reads as an answer: %r" % sorted(out3)) finally: globals()["_post"] = _real_post3 for _miss in ({"groesse": "klein", "stichtag": "2025-12-31"}, {"rechtsform": "GmbH", "stichtag": "2025-12-31"}): try: _prepare(dict(_miss, wirtschaftsjahr="2025")) fails.append("a call missing %s was forwarded — the engine would default it" % ("rechtsform" if "rechtsform" not in _miss else "groesse")) except ValueError: pass for _t in TOOLS: if _t["fn"] is t_eligibility: _req = set(_t["inputSchema"].get("required") or []) if not {"rechtsform", "groesse", "stichtag"} <= _req: fails.append("ebf_check_eligibility does not require what the engine defaults: %s" % _req) try: _prepare({"rechtsform": "GmbH", "groesse": "klein", "wirtschaftsjahr": "2025"}) fails.append("a missing stichtag was forwarded — the engine answers eligible=false for it") except ValueError: pass _ok = _prepare({"rechtsform": "gmbh", "groesse": "kleinst", "wirtschaftsjahr": "2025", "stichtag": "2025-12-31"}) if _ok != {"legal_form": "GmbH", "size_class": "Kleinst", "fy": 2025, "stichtag": "2025-12-31"}: fails.append("_prepare did not canonicalise onto the engine's spelling: %r" % _ok) for _t in TOOLS: if _t["fn"] in (t_eligibility, t_validate): if "stichtag" not in (_t["inputSchema"].get("required") or []): fails.append("%s does not declare stichtag required" % _t["name"]) for p_ in _annotation_problems(TOOLS): fails.append("annotations: " + p_) if not _annotation_problems([{"name": "planted", "description": "never files"}]): fails.append("the annotation check passed a tool with no title and no annotations — it cannot fail") if not _annotation_problems([{"name": "planted", "title": "t", "annotations": dict(_READ_ONLY, destructiveHint=True)}]): fails.append("the annotation check passed a tool that claims to be destructive") _ro = {t["name"]: (t.get("annotations") or {}).get("readOnlyHint") for t in TOOLS} for _n, _want in (("ebf_check_eligibility", False), ("ebf_validate_figures", False), ("ebf_get_products", True), ("ebf_get_pricing", True), ("ebf_build_filing_link", True)): if _ro.get(_n) is not _want: fails.append("%s readOnlyHint is %r, want %r" % (_n, _ro.get(_n), _want)) _listed = handle({"jsonrpc": "2.0", "id": 50, "method": "tools/list"})["result"]["tools"] if any("fn" in t or "annotations" not in t or "title" not in t for t in _listed): fails.append("tools/list leaks `fn` or drops title/annotations: %r" % [sorted(t) for t in _listed]) for _asked, _want in (("2025-06-18", "2025-06-18"), ("2024-11-05", "2024-11-05"), ("1999-01-01", PROTOCOL_VERSIONS[0]), (None, PROTOCOL_VERSIONS[0])): _params = {} if _asked is None else {"protocolVersion": _asked} _got = handle({"jsonrpc": "2.0", "id": 51, "method": "initialize", "params": _params})["result"]["protocolVersion"] if _got != _want: fails.append("initialize asked %r answered %r, want %r" % (_asked, _got, _want)) _remote = [t for t in TOOLS if t["name"] in REMOTE_TOOL_NAMES] if sorted(t["name"] for t in _remote) != sorted(REMOTE_TOOL_NAMES): fails.append("REMOTE_TOOL_NAMES names a tool that does not exist: %r" % (REMOTE_TOOL_NAMES,)) _rl = handle({"jsonrpc": "2.0", "id": 52, "method": "tools/list"}, tools=_remote)["result"]["tools"] if sorted(t["name"] for t in _rl) != sorted(REMOTE_TOOL_NAMES): fails.append("the remote tools/list is not the remote set: %r" % [t["name"] for t in _rl]) for _blocked in ("ebf_validate_figures", "ebf_build_filing_link"): _r = handle({"jsonrpc": "2.0", "id": 53, "method": "tools/call", "params": {"name": _blocked, "arguments": {}}}, tools=_remote) if (_r.get("error") or {}).get("code") != -32601: fails.append("%s was callable through the remote set: %r" % (_blocked, _r)) _reached = handle({"jsonrpc": "2.0", "id": 54, "method": "tools/call", "params": {"name": "ebf_check_eligibility", "arguments": {}}}, tools=_remote) if "error" in _reached or not (_reached.get("result") or {}).get("isError"): fails.append("a listed remote tool was not reached (its missing-field refusal should come back as content): %r" % _reached) print(" self-test: protocol (initialize · notification · tools/list · unknown tool) + the " "filing boundary asserted on all %d tool descriptions + no tool accepts a credential + " "the link round-trips and refuses 3 malformed inputs, and the engine field mapping " "holds (incl. fy as int and a misspelling refused). DW-1364: token case-folding with " "an unknown-token control, a 422 labelled AS a refusal at the `_post` level with a " "genuine-answer control, the refused-validate shape, the required Stichtag. DW-1368: a " "200 carrying no judgement is caught, with a real verdict AND a legible out-of-scope as " "the two permit halves, plus the two fields the engine silently defaults. " "(No case COUNT is printed here on purpose — the previous line claimed 12 and matched " "no count of what actually runs, which is prose read as evidence.)" % len(TOOLS)) for f in fails: print(" x " + f) print("OK self-test passed" if not fails else "X self-test FAILED") return 1 if fails else 0 if __name__ == "__main__": if "--self-test" in sys.argv: raise SystemExit(self_test()) raise SystemExit(serve())