#!/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 expressly does not take it on. Where the law allows more than one mapping, the tool does not choose for the customer; the customer confirms the mapping and 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 #: Stated once, here, and repeated in every tool's description that could be misread as an offer to #: file. An agent reads descriptions, so the boundary has to live where the agent looks. BOUNDARY = ("This server never files, never takes payment, and never accepts an ELSTER certificate " "or PIN. It prepares; the human reviews, pays and signs.") 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: # a refusal is an ANSWER, not a crash # DW-1364 — AND IT MUST ARRIVE LABELLED AS ONE. This used to return the error BODY as if it # were a result, so `res.get("verdict")` was None, `checks` was {} and `checks_failed` was # [] — a 422 reached the agent as "verdict null, nothing failed, no reasons". Driven live: # sending numeric Bilanz values (which the schema invited) returned HTTP 422 and the tool # rendered a silent empty pass. A rejection is a routing decision; this is where it routed. try: detail = json.loads(e.read().decode("utf-8")) except Exception: # noqa: BLE001 detail = None return {"_refused": True, "_http": int(e.code), "_detail": detail} #: The tokens the ENGINE accepts, declared once and used for BOTH the schema an agent reads and the #: canonicaliser below — a second list would drift from the first, which is the defect this file's #: header already names about prices. #: DW-1373 — a COPY of `ebilanz_fabrik_engine.eligibility.LEGAL_FORMS` plus `andere`, and a copy #: is exactly what drifts: the first cut carried 4 of the engine's 7 accepted spellings, so an agent #: sending "aktiengesellschaft" or "ug (haftungsbeschränkt)" was never case-folded and the engine #: refused a form it serves. This file is zero-deps (it runs on a stranger's laptop), so it cannot #: import the engine — `code/tests/test_dw1373_the_mcp_token_lists_cover_the_engine.py` compares the #: two instead, so the next widening of the engine cannot ship without this list following it. #: `andere` is deliberately NOT an engine token: it is how an agent says "none of these", and the #: engine answers it with a legible out-of-scope reason rather than a silent default. _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} 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} # ── the tools ─────────────────────────────────────────────────────────────────────────────────── 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: # noqa: BLE001 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."} #: The engine's own field names, mapped from the ones an agent would naturally reach for. #: EXPLICIT, and unknown keys are REFUSED rather than passed through — `AmpelCheckRequest` is a #: pydantic model, so an undeclared key is SILENTLY DROPPED (its own DW-774 comment says exactly #: that). Driven live while building this: sending `wirtschaftsjahr` got back "Bitte geben Sie das #: Wirtschaftsjahr an" — the value vanished and the agent was handed a refusal about the field it #: had just supplied. A wrong name must be an error here, not a disappearance there. _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) # THE STICHTAG IS LOAD-BEARING AND THE SCHEMA USED TO CALL IT OPTIONAL. Omitted, the engine # answers `Stichtag '' is not 31 December` and returns `eligible: false` — a company we serve, # told no, because of a field the agent was told it could leave out. Refuse here instead, where # the message can say what to do. # DW-1368b — AND SO ARE THE TWO FIELDS THE ENGINE SILENTLY DEFAULTS. `AmpelCheckRequest` # declares `legal_form="GmbH"` and `size_class="Kleinst"` as DEFAULTS, so omitting them does not # produce a refusal — it produces an answer about a company nobody described. Measured live: # `ebf_check_eligibility` with only a Wirtschaftsjahr and a Stichtag replied `eligible: true`. # An agent relays that as "you can be served"; for a Limited or a mittelgrosse AG it is false. 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") 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, "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 {} # A NULL CHECK IS ONE THAT WAS NOT TAKEN, NEVER ONE THAT PASSED. The engine stops at the first # structural failure, so everything after it comes back `null` — and an agent reading a JSON # `null` as "fine" would tell its person the balance is in order when nothing ever looked at it. # Named here because the agent reads this object, not our source. 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, "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, "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, "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, "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": { # DW-1364 — the engine's model is `dict[str, str]`. A NUMBER here is rejected with HTTP 422 # ("Input should be a valid string"), which this server used to render as an empty pass. # Say string, and say it in the example. "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, "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} # ── JSON-RPC 2.0 over stdio ───────────────────────────────────────────────────────────────────── def _result(rid, payload): return {"jsonrpc": "2.0", "id": rid, "result": payload} def handle(msg: dict) -> dict | None: m, rid = msg.get("method"), msg.get("id") # DW-1372 — A NOTIFICATION NEVER GETS A REPLY, WHATEVER ITS METHOD. JSON-RPC 2.0 § 4.1: a # request without an `id` member is a notification and "the Server MUST NOT reply". This used # to special-case `notifications/initialized` alone, so `notifications/cancelled` — which a # client sends the moment a user aborts a slow call — came back as an error with `id: null`, # a reply to a message that asked for none. Decided on the ABSENCE of the key, not on the # method name: the method list is a vocabulary, "has no id" is the invariant. if "id" not in msg: return None if m == "ping": # MCP basic/utilities: a ping is answered with an EMPTY result. Clients use it as a # liveness check; -32601 read as "server broken". return _result(rid, {}) if m == "initialize": return _result(rid, {"protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, "serverInfo": {"name": "ebilanz-fabrik", "version": "1.0.0"}}) if m in ("notifications/initialized", "initialized"): return None # a notification takes no reply if m == "tools/list": return _result(rid, {"tools": [{k: t[k] for k in ("name", "description", "inputSchema")} 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: # noqa: BLE001 # A failure is reported as tool CONTENT with isError, never as a silent empty result — # an agent that cannot tell "it failed" from "nothing to report" will report nothing. 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: # noqa: BLE001 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 = [] # ── protocol, offline ── 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") # ── THE BOUNDARY IS PART OF THE CONTRACT, so it is asserted, not trusted ── 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)) # ── the link builder, offline and two-sided ── 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 # ── the field mapping is a CONTRACT with the engine, so it is asserted ── 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"}) # a typo an agent will make fails.append("a misspelled field was forwarded — pydantic would drop it silently") except ValueError: pass # ── DW-1364: the three defects driven against production, each with its control ────────── # (a) the token case. `Kleinst` was served and `kleinst` refused, in a sentence that said we # cover Kleinst. The control that matters is the THIRD one: an unknown form must NOT be # snapped onto a real one, or this helper would invent an eligibility. 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)) # (b) a refusal must be legible AS a refusal, and a real answer must not be mistaken for one. 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") # (b1) …and `_post` ITSELF must produce that label. The case below patches `_post` away, so on # its own it proves nothing about the function the tools actually call — the mutation control # caught exactly that: deleting the label from `_post` left the suite green. Drive the real one. 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 # (b2) THE NEGATIVE CONTROL FOR THE DEFECT ITSELF: with the engine refusing, `t_validate` must # not hand back an object whose every failure field is empty. That is exactly what it did. _real_post = globals()["_post"] try: globals()["_post"] = lambda *_a, **_k: {"_refused": True, "_http": 422, "_detail": {"d": 1}} # carries legal_form/size_class because DW-1368b made them required — the engine DEFAULTS # them, so a fixture that omits them is asking a different question than it looks like. 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 # (f) DW-1372 — JSON-RPC notifications and MCP ping. Two-sided on purpose: swallowing every # unknown method would also swallow the error a REAL request is owed. 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") # (d) DW-1368 — a 200 that is not a judgement. `_post` returns whatever the engine sends; only # an HTTPError carries `_refused`. These three shapes arrive as 200 and must not read as # answers. The permit halves matter as much: a real verdict and a legible out-of-scope both # have to pass straight through, or the tool would refuse to ever answer anything. 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 # (e) DW-1368b — the engine DEFAULTS legal_form and size_class, so omitting them answers a # different question rather than refusing. Measured: fy+stichtag alone -> eligible: true. 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) # (c) the Stichtag is load-bearing, so it is refused HERE rather than silently downgraded to an # out-of-scope verdict by the engine. 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"]) 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())