"""Threaded publish-path sequence (Example: Publishing), runnable. Runs steps 0-2 of the article against a local MCP server spawned via stdio, threading the request and replayControl objects straight from the step-0 response into the capture and prepare calls — as objects, never retyped. Prints a compact report and saves the raw transcript. Expected on an unconfigured answers service: step 0 COMPUTED / TRUE_ONLY; steps 1-2 refuse with ANSWERS_UNAVAILABLE. Steps 3-4 (publish, revoke) need a preparation id and stay NOT RUN. Run from the repo root: LAW_MCP_PROFILE=kz python3 docs/agent-engineering/examples/publishing/thread_sequence.py [--out transcript.json] Requires the prebuilt server binary at engines/lawc/target/debug/law-mcp-server. Standard library only. """ import json import os import subprocess import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[4] SERVER = ROOT / "engines" / "lawc" / "target" / "debug" / "law-mcp-server" ASK_ARGS = { "kind": "truth", "package": "kz-labour-code", "predicate": "feeding_break_too_short", "args": ["urn:kz:tk:employee:1", "urn:kz:tk:employer:1"], "facts": [ { "predicate": "children_under_eighteen_months", "args": ["urn:kz:tk:employee:1", 1], }, { "predicate": "child_feeding_break_minutes", "args": ["urn:kz:tk:employee:1", "urn:kz:tk:employer:1", 25], }, ], "legalTime": "2026-09-01", } class Server: def __init__(self): env = dict(os.environ) env.setdefault("LAW_MCP_PROFILE", "kz") self.proc = subprocess.Popen( [str(SERVER)], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, bufsize=1, cwd=str(ROOT), env=env, ) self.next_id = 1 def call(self, method, params=None): payload = {"jsonrpc": "2.0", "id": self.next_id, "method": method} if params is not None: payload["params"] = params self.next_id += 1 self.proc.stdin.write(json.dumps(payload) + "\n") self.proc.stdin.flush() line = self.proc.stdout.readline() if not line: raise RuntimeError("server closed stdout") return json.loads(line) def tool(self, name, arguments): return self.call( "tools/call", {"name": name, "arguments": arguments} ) def close(self): self.proc.stdin.close() self.proc.wait(timeout=30) def short(text, n=120): text = " ".join(str(text).split()) return text if len(text) <= n else text[:n] + "…" def steps34_message(step2): """Explain steps 3-4 from the actual prepare response. The script never calls publish or revoke; the reason it stops depends on what prepare returned: a refusal (quote it), a real preparation id (publish/revoke simply not attempted here), or a success-shaped response the script does not recognize. """ result = step2.get("result", {}) if result.get("isError"): return ( "steps 3-4 (publish, revoke): NOT RUN — prepare refused " "(%s)" % short(result.get("structuredContent", result.get("content"))) ) pid = (result.get("structuredContent") or {}).get("preparationId") if pid: return ( "steps 3-4 (publish, revoke): NOT RUN — this script stops " "at prepare (preparation id %s issued; publish/revoke not " "attempted)" % pid ) return ( "steps 3-4 (publish, revoke): NOT RUN — prepare returned no " "error but no preparation id (unrecognized shape)" ) def main(argv): out_path = None args = list(argv) if "--out" in args: i = args.index("--out") out_path = Path(args[i + 1]) srv = Server() transcript = [] try: init = srv.call("initialize", { "protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "ae-publish-thread", "version": "0.1.0"}, }) srv.proc.stdin.write(json.dumps( {"jsonrpc": "2.0", "method": "notifications/initialized"} ) + "\n") srv.proc.stdin.flush() info = init.get("result", {}).get("serverInfo", {}) print(f"server: {info.get('name')} {info.get('version')}") # Step 0 — ask. The request echo and replayControl below are # taken from this response as objects. step0 = srv.tool("law_ask", ASK_ARGS) transcript.append({"step": 0, "tool": "law_ask", "request": ASK_ARGS, "response": step0}) sc0 = step0["result"]["structuredContent"] answer = sc0["answer"] control = sc0["replayControl"] assert control["schemaVersion"] == "law.answers.replay-control/0.1" assert control["request"]["predicate"] == ASK_ARGS["predicate"] print(f"step 0 ask: {answer['evaluationStatus']} / " f"{answer['truthStatus']}; " f"resultHash {control['checks']['resultHash'][:23]}…") # Step 1 — capture threads the echoed request object. capture_args = {"producer": "law_ask", "request": control["request"]} step1 = srv.tool("law_capture_answer", capture_args) transcript.append({"step": 1, "tool": "law_capture_answer", "request": capture_args, "response": step1}) print(f"step 1 capture: isError={step1['result'].get('isError')} " f"{short(step1['result'].get('structuredContent', step1['result'].get('content')))}") # Step 2 — prepare threads request + replayControl objects. prepare_args = { "question": "Is the feeding break too short?", "calculations": [{ "id": "c1", "label": "feeding break truth", "request": control["request"], "replayControl": control, }], } step2 = srv.tool("law_prepare_answer", prepare_args) transcript.append({"step": 2, "tool": "law_prepare_answer", "request": prepare_args, "response": step2}) print(f"step 2 prepare: isError={step2['result'].get('isError')} " f"{short(step2['result'].get('structuredContent', step2['result'].get('content')))}") finally: srv.close() if out_path is not None: out_path.write_text( json.dumps(transcript, ensure_ascii=False, indent=1) + "\n") print(f"transcript: {out_path} " f"({sum(len(json.dumps(t)) for t in transcript)} bytes)") print(steps34_message(step2)) return 0 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))