Compare paired runs of the same inputs and scorer. A returned value alone is insufficient: the example independently checks the invoice total before recording quality evidence.
When to use this
Use this when deciding whether a changed implementation improves a fixed workload at equivalent verified quality.
Install the matching API preview
These examples use the API 2 preview, not names promised by the public RC5 release. Existing RC5 entry points remain available. The Python wheel below is verified on Windows x64; it is not a cross-platform release.
python -m pip install https://docs.soterias.io/downloads/api-preview/soterias_local-0.1.0rc5%2Bapi2-cp311-abi3-win_amd64.whlnpm install https://docs.soterias.io/downloads/api-preview/soterias-core-0.1.0-rc.5-api.2.tgz
npm install --save-dev typescript@5.8.3 @types/node@24Python imports soterias. TypeScript imports @soterias/core. Use Python 3.11+ for this wheel and Node 22.23.4+ for these examples. For a .mts file, compile with npx tsc example.mts --target ES2022 --module NodeNext --moduleResolution NodeNext --skipLibCheck, then run the emitted .mjs file.
Before you run
A fresh output directory and your benchmark provenance manifest. The optional benchmark branch records both variants, five matched pairs, input/scorer identities and explicit quality evidence. Without that manifest the example deliberately produces no comparison.
Application walkthrough
Choose Python or TypeScript. Create example.py or example.mts and append that language's chunks in order. Each language forms one complete program; indentation and open blocks carry across chunks.
Import the packages
Import the SDK and standard-library helpers. These are the real package exports used by the executable program.
"""Installed native local metrics, real CSV accounting, refusal and optional fair raw-run benchmark."""
from __future__ import annotations
import argparse
import asyncio
import csv
import hashlib
import io
import json
import os
from pathlib import Path
import platform
import sqlite3
import sys
import time
from soterias import Metrics, MetricsError, canonicalize, metrics_cohort_digest, metrics_run_schema
def digest(data):
return hashlib.sha256(data).hexdigest()
def encode(value):
return canonicalize(json.dumps(value, separators=(",", ":")).encode())// Actual installed metrics, real CSV accounting and explicit local baseline comparison.
import assert from 'node:assert/strict';
import fs from 'node:fs';
import path from 'node:path';
import crypto from 'node:crypto';
import os from 'node:os';
import {DatabaseSync} from 'node:sqlite';
import {performance} from 'node:perf_hooks';
import {fileURLToPath} from 'node:url';
import {Metrics, MetricsError, canonicalize, metricsCohortDigest, metricsRunSchema, type MetricsComparisonData} from '@soterias/core';
const [argument,...rest]=process.argv.slice(2);
assert.ok(argument&&rest.length<=1&&(rest.length===0||rest[0]!.startsWith('--benchmark-manifest=')),'Usage: accounting.mjs NEW_DIRECTORY [--benchmark-manifest=FILE]');
const root=path.resolve(argument);fs.mkdirSync(root,{mode:0o700});
const hash=(value:Uint8Array)=>crypto.createHash('sha256').update(value).digest('hex');
const encode=(value:unknown)=>canonicalize(Buffer.from(JSON.stringify(value))).canonical;
function write(name:string,data:Uint8Array){const descriptor=fs.openSync(path.join(root,name),'wx',0o600);try{fs.writeFileSync(descriptor,data);fs.fsyncSync(descriptor);}finally{fs.closeSync(descriptor);}}
write('invoices.csv',Buffer.from('invoice,amount_cents\nUNIT-1,17500\nUNIT-2,24000\nUNIT-3,8500\n'));
let reads=0,returned:{total_cents:number;rows:number}|undefined;
function existingWorkflow(){const rows=fs.readFileSync(path.join(root,'invoices.csv'),'utf8').trimEnd().split('\n').slice(1);reads++;returned={total_cents:rows.reduce((total,row)=>total+Number(row.split(',')[1]),0),rows:rows.length};return returned;}
function verify(value:unknown){assert.deepEqual(value,{total_cents:50000,rows:3});}
// Three ordinary integration lines including the import above; no policy/daemon/helper setup.
const metrics=new Metrics({workflow:'accounting',variant:'instrumented',directory:path.join(root,'metrics')});
const tracked=metrics.track(existingWorkflow);Step 2
Create the local export explicitly. The returned bytes remain local unless your application chooses to send them elsewhere.
def write(path, data):
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(descriptor, "wb") as output:
output.write(data)
output.flush()
os.fsync(output.fileno())
async def main(root, manifest):
root.mkdir(mode=0o700)
business = root / "invoices.csv"
write(business, b"invoice,amount_cents\nUNIT-1,17500\nUNIT-2,24000\nUNIT-3,8500\n")
reads, returned = 0, None
def existing_workflow():
nonlocal reads, returned
with business.open(newline="", encoding="utf-8") as source:
rows = list(csv.DictReader(source))
reads += 1
returned = {"total_cents": sum(int(row["amount_cents"]) for row in rows), "rows": len(rows)}
return returnedfor(let index=0;index<5;index++){const value=tracked();assert.equal(value,returned);verify(value);}
const trackedAsync=metrics.trackAsync(async()=>{await Promise.resolve();return existingWorkflow();});
for(let index=0;index<5;index++){const value=await trackedAsync();assert.equal(value,returned);verify(value);}
const originalError=new Error('disposable UNIT original application error');
assert.throws(()=>metrics.track(()=>{throw originalError;})(),error=>error===originalError);
await assert.rejects(metrics.trackAsync(async()=>{throw originalError;})(),error=>error===originalError);
let entered!:()=>void;const started=new Promise<void>(resolve=>{entered=resolve;});
const controller=new AbortController(),reason=new DOMException('disposable UNIT application cancellation','AbortError');
const cancelled=metrics.trackAsync(async()=>{entered();await new Promise<void>((resolve,reject)=>{controller.signal.addEventListener('abort',()=>reject(controller.signal.reason),{once:true});});})();
await started;controller.abort(reason);await assert.rejects(cancelled,error=>error===reason);await metrics.flush();
assert.equal(metrics.pendingRecordings,0);assert.equal(metrics.lastRecordingError,null);
let comparison:MetricsComparisonData|null=null;
if(rest.length){
const provenance=JSON.parse(fs.readFileSync(rest[0]!.slice('--benchmark-manifest='.length),'utf8'));assert.equal(provenance.version,1);
const program=fs.readFileSync(fileURLToPath(import.meta.url)),scorer=hash(Buffer.concat([Buffer.from('fixed-local-accounting-scorer-v1\0'),program]));
const input=hash(fs.readFileSync(path.join(root,'invoices.csv')));
const cohort={version:1,workflow:'accounting',input_distribution:input,scorer,git_sha:provenance.git_sha,git_dirty:provenance.git_dirty,
code:hash(Buffer.concat([program,Buffer.from(provenance.source_snapshot_sha256,'hex')])),package_builds:provenance.package_builds,
benchmark:'installed-local-accounting-metrics-v1',dataset_version:'synthetic-invoices-v1',dataset_license:'CC0',held_out_split:'fixed-local-case',
environment:{os:process.platform,architecture:os.arch(),runtime:'Node-'+process.versions.node,hardware:provenance.hardware},
cache:'warm',timing_scope:'callable-and-recording-excludes-scorer-and-collector',policy:null,graph:null,model:null,pricing_source_version:null};
const cohortHash=metricsCohortDigest(encode(cohort)),collector=new Metrics({workflow:'accounting',variant:'instrumented',directory:path.join(root,'benchmark-metrics')});
for(let replicate=0;replicate<5;replicate++)for(const variant of(replicate%2===0?['baseline','instrumented']:['instrumented','baseline'])){Step 3
Check the result against independent evidence and retain refusals. The assertion makes this example fail if native verification behaves differently.
def verify(value):
# Independent fixed business criterion, not a claim that callback return implies success.
assert value == {"total_cents": 50_000, "rows": 3}
# Three ordinary integration lines including the import above; no policy/server/helper setup.
metrics = Metrics("accounting", variant="instrumented", directory=root / "metrics")
tracked = metrics.track(existing_workflow)
for _ in range(5):
value = tracked()
assert value is returned
verify(value)
async def existing_async():
await asyncio.sleep(0)
return existing_workflow()
tracked_async = metrics.track_async(existing_async)
for _ in range(5):
value = await tracked_async()
assert value is returned
verify(value)
original_error = RuntimeError("disposable UNIT application error")
def failed():
raise original_error
try:
metrics.track(failed)() const begin=Date.now(),clock=performance.now(),value=variant==='baseline'?existingWorkflow():tracked(),elapsed=Math.round((performance.now()-clock)*1000000),end=Date.now();verify(value);
const unmeasured=()=>({kind:'unmeasured'}),usage:Record<string,unknown>={};
for(const name of['input_tokens','output_tokens','cached_input_tokens','model_calls','tool_calls','cache_hits','graph_hits','graph_qualified','invalidations','fallback_count','provider_cost'])usage[name]=unmeasured();usage.rates=null;
const overhead:Record<string,unknown>={soterias_ns:unmeasured(),policy_ns:unmeasured(),broker_ns:unmeasured(),approval_ns:unmeasured()};
if(variant==='baseline')overhead.soterias_ns={kind:'observed',value:'0',source:{kind:'caller_reported'}};
else{const inner=BigInt(metrics.lastObservation!.data().run.duration_ns);if(BigInt(elapsed)>=inner)overhead.soterias_ns={kind:'observed',value:String(BigInt(elapsed)-inner),source:{kind:'caller_reported'}};}
collector.append(encode({version:1,id:`${variant}-${replicate}`,workflow:'accounting',variant,cohort,input,replicate:String(replicate),warmup:false,
started_at_ms:String(begin),ended_at_ms:String(end),duration_ns:String(elapsed),timing_source:{kind:'caller_reported'},outcome:'returned',
quality:{kind:'verified',verifier:scorer,output:hash(encode(value)),score_basis_points:'10000',evidence:hash(encode({criterion:'fixed-total-and-row-count',actual:value}))},
usage,overhead,warnings:['fixed-local-workload','os-cache-not-isolated','no-provider-usage','uncertainty-not-estimated']}));
}
comparison=collector.compare(cohortHash,'baseline','instrumented').data();assert.equal(comparison.matched_pairs,5);assert.equal(comparison.failed_outcome_pairs,0);assert.equal(comparison.baseline.counts.verified,5);assert.equal(comparison.variant.counts.verified,5);assert.equal(comparison.baseline.paired_duration_ns!.samples,'5');
const excluded=metrics.compare(cohortHash,'baseline','instrumented').data();assert.equal(excluded.matched_pairs,0);assert.equal(excluded.unscored_store_runs,18);
write('benchmark-comparison.json',encode(comparison));write('benchmark-raw.json',collector.exportJson().data);write('benchmark-raw.csv',collector.exportCsv().data);collector.close();
}
const expected=rest.length?18:13,summary=metrics.snapshot().data();
assert.equal(summary.counts.raw,expected);assert.equal(summary.counts.returned,expected-3);assert.equal(summary.counts.error,2);assert.equal(summary.counts.cancelled,1);assert.equal(summary.counts.unverified,expected);assert.equal(summary.paired_duration_ns,null);
assert.equal(summary.all_runs.duration_ns.buckets[0]!.distribution.samples,String(expected));assert.deepEqual(summary.all_runs.input_tokens.buckets,[]);assert.deepEqual(summary.all_cost.buckets,[]);
const exported=metrics.exportJson(),records=JSON.parse(exported.data.toString('utf8')).records;assert.equal(exported.total,expected);assert.equal(exported.nextOffset,null);
for(const record of records){assert.equal(record.cohort,null);assert.equal(record.input,null);assert.equal(record.replicate,null);assert.deepEqual(record.usage.input_tokens,{kind:'unmeasured'});}
const csv=metrics.exportCsv().data;assert.ok(csv.toString('utf8').startsWith('run_id,workflow,cohort,variant,input,replicate,duration_ns,raw_record_json\r\n'));
write('observations.json',exported.data);write('observations.csv',csv);
const changed=metrics.snapshot().data() as any;changed.counts.raw=9999;assert.equal(metrics.snapshot().data().counts.raw,expected);Step 4
Exercise cancellation without rewriting the original application outcome. Native work that already started may still finish.
except RuntimeError as observed:
assert observed is original_error
else:
raise AssertionError("original synchronous error was rewritten")
async def failed_async():
raise original_error
try:
await metrics.track_async(failed_async)()
except RuntimeError as observed:
assert observed is original_error
else:
raise AssertionError("original asynchronous error was rewritten")
entered, wait = asyncio.Event(), asyncio.Event()
async def cancellable():
entered.set()
await wait.wait()
cancelled = asyncio.create_task(metrics.track_async(cancellable)())
await entered.wait()
cancelled.cancel()
try:
await cancelled
except asyncio.CancelledError:
pass
else:const originalBytes=metrics.snapshot().canonicalBytes(),copy=metrics.snapshot().canonicalBytes();copy.fill(0);assert.notDeepEqual(copy,originalBytes);assert.equal(metrics.snapshot().data().counts.raw,expected);
assert.ok(JSON.parse(metricsRunSchema().toString('utf8')).properties.cohort.anyOf);await metrics.closeAsync();
const reopened=new Metrics({workflow:'accounting',variant:'instrumented',directory:path.join(root,'metrics')});assert.equal(reopened.snapshot().data().counts.raw,expected);reopened.close();
// Actual disposable native-schema corruption; optional recording failure preserves the callback's value.
const broken=new Metrics({workflow:'failure-case',directory:path.join(root,'broken-metrics')});
const database=new DatabaseSync(path.join(root,'broken-metrics/metrics.sqlite3'));try{database.exec('CREATE TABLE unexpected(value TEXT)');}finally{database.close();}
const marker={},count={calls:0};assert.equal(broken.track(()=>{count.calls++;return marker;})(),marker);assert.equal(count.calls,1);assert.ok(broken.lastRecordingError instanceof MetricsError);assert.equal(broken.lastRecordingError.code,'metrics_store_corrupt');assert.equal(broken.pendingRecordings,0);broken.close();
console.log(JSON.stringify({status:'PASS_INSTALLED_NATIVE_LOCAL_METRICS_COMPONENT',language:'typescript',actualBusinessFileReads:reads,persistedOrdinaryObservations:expected,
unscoredObservationsNotCompared:true,usageAndCostUnmeasured:true,originalValueErrorAndCancellationPreserved:true,recordingFailureDidNotChangeApplication:true,
csvJsonExported:true,reopenedSameNativeStore:true,benchmark:comparison,networkSyscallsAudited:false,providerUsageExecuted:false,securityEnforcement:false,fullTask27Complete:false}));Step 5
Exercise cancellation without rewriting the original application outcome. Native work that already started may still finish.
raise AssertionError("application cancellation became success")
await metrics.flush()
assert metrics.pending_recordings == 0 and metrics.last_recording_error is None
comparison = None
if manifest is not None:
provenance = json.loads(manifest.read_bytes())
assert provenance["version"] == 1
program = Path(__file__).read_bytes()
scorer = digest(b"fixed-local-accounting-scorer-v1\0" + program)
input_hash = digest(business.read_bytes())
cohort = {"version": 1, "workflow": "accounting", "input_distribution": input_hash, "scorer": scorer,
"git_sha": provenance["git_sha"], "git_dirty": provenance["git_dirty"],
"code": digest(program + bytes.fromhex(provenance["source_snapshot_sha256"])),
"package_builds": provenance["package_builds"], "benchmark": "installed-local-accounting-metrics-v1",
"dataset_version": "synthetic-invoices-v1", "dataset_license": "CC0", "held_out_split": "fixed-local-case",
"environment": {"os": sys.platform, "architecture": platform.machine(),
"runtime": "CPython-" + platform.python_version(), "hardware": provenance["hardware"]},
"cache": "warm", "timing_scope": "callable-and-recording-excludes-scorer-and-collector",
"policy": None, "graph": None, "model": None, "pricing_source_version": None}
cohort_hash = metrics_cohort_digest(encode(cohort))
collector = Metrics("accounting", variant="instrumented", directory=root / "benchmark-metrics")
for replicate in range(5):
# Alternate order after ten actual warmup reads; neither variant's sample is hidden.Step 6
Check the result against independent evidence and retain refusals. The assertion makes this example fail if native verification behaves differently.
for variant in (["baseline", "instrumented"] if replicate % 2 == 0 else ["instrumented", "baseline"]):
started = time.time_ns() // 1_000_000
clock = time.perf_counter_ns()
result = existing_workflow() if variant == "baseline" else tracked()
elapsed = time.perf_counter_ns() - clock
ended = time.time_ns() // 1_000_000
verify(result)
unmeasured = lambda: {"kind": "unmeasured"}
usage = {name: unmeasured() for name in ("input_tokens", "output_tokens", "cached_input_tokens", "model_calls",
"tool_calls", "cache_hits", "graph_hits", "graph_qualified", "invalidations", "fallback_count", "provider_cost")}
usage["rates"] = None
overhead = {name: unmeasured() for name in ("soterias_ns", "policy_ns", "broker_ns", "approval_ns")}
if variant == "baseline":
overhead["soterias_ns"] = {"kind": "observed", "value": "0", "source": {"kind": "caller_reported"}}
else:
inner = int(metrics.last_observation.data()["run"]["duration_ns"])
if elapsed >= inner:
overhead["soterias_ns"] = {"kind": "observed", "value": str(elapsed - inner), "source": {"kind": "caller_reported"}}
raw = {"version": 1, "id": f"{variant}-{replicate}", "workflow": "accounting", "variant": variant,
"cohort": cohort, "input": input_hash, "replicate": str(replicate), "warmup": False,
"started_at_ms": str(started), "ended_at_ms": str(ended), "duration_ns": str(elapsed),
"timing_source": {"kind": "caller_reported"}, "outcome": "returned",
"quality": {"kind": "verified", "verifier": scorer, "output": digest(encode(result)),
"score_basis_points": "10000", "evidence": digest(encode({"criterion": "fixed-total-and-row-count", "actual": result}))},Step 7
Check the result against independent evidence and retain refusals. The assertion makes this example fail if native verification behaves differently.
"usage": usage, "overhead": overhead,
"warnings": ["fixed-local-workload", "os-cache-not-isolated", "no-provider-usage", "uncertainty-not-estimated"]}
collector.append(encode(raw))
comparison = collector.compare(cohort_hash, "baseline", "instrumented").data()
assert comparison["matched_pairs"] == 5 and comparison["failed_outcome_pairs"] == 0
assert comparison["baseline"]["counts"]["verified"] == 5 and comparison["variant"]["counts"]["verified"] == 5
assert comparison["baseline"]["paired_duration_ns"]["samples"] == "5"
# The primary ordinary observations remain ineligible despite identical workflow labels.
rejected = metrics.compare(cohort_hash, "baseline", "instrumented").data()
assert rejected["matched_pairs"] == 0 and rejected["unscored_store_runs"] == 18
write(root / "benchmark-comparison.json", encode(comparison))
write(root / "benchmark-raw.json", collector.export_json().data)
write(root / "benchmark-raw.csv", collector.export_csv().data)
collector.close()
expected = 18 if manifest is not None else 13
summary = metrics.snapshot().data()
assert summary["counts"]["raw"] == expected and summary["counts"]["returned"] == expected - 3
assert summary["counts"]["error"] == 2 and summary["counts"]["cancelled"] == 1
assert summary["counts"]["unverified"] == expected and summary["paired_duration_ns"] is None
assert summary["all_runs"]["duration_ns"]["buckets"][0]["distribution"]["samples"] == str(expected)
assert summary["all_runs"]["input_tokens"]["buckets"] == [] and summary["all_cost"]["buckets"] == []
export = metrics.export_json()
original_rows = json.loads(export.data)["records"]Step 8
Pair only compatible runs with the same input and scoring evidence. Record both variants instead of comparing unrelated aggregate timings.
assert export.total == expected and export.next_offset is None
assert all(run["cohort"] is None and run["input"] is None and run["replicate"] is None for run in original_rows)
assert all(run["usage"]["input_tokens"] == {"kind": "unmeasured"} for run in original_rows)
csv_page = metrics.export_csv()
roundtrip = [json.loads(row["raw_record_json"]) for row in csv.DictReader(io.StringIO(csv_page.data.decode()))]
assert roundtrip == original_rows
write(root / "observations.json", export.data)
write(root / "observations.csv", csv_page.data)
separate = metrics.snapshot().data()
separate["counts"]["raw"] = 9999
assert metrics.snapshot().data()["counts"]["raw"] == expected
schema = metrics_run_schema()
assert json.loads(schema)["properties"]["cohort"]["anyOf"]
await metrics.close_async()
reopened = Metrics("accounting", variant="instrumented", directory=root / "metrics")
assert reopened.snapshot().data()["counts"]["raw"] == expected
reopened.close()Step 9
Create the local export explicitly. The returned bytes remain local unless your application chooses to send them elsewhere.
# Real disposable schema corruption, not a mock backend: observability failure cannot replace a value.
broken = Metrics("failure-case", directory=root / "broken-metrics")
database = sqlite3.connect(root / "broken-metrics/metrics.sqlite3")
try:
database.execute("CREATE TABLE unexpected(value TEXT)")
database.commit()
finally:
database.close()
marker, calls = object(), 0
def unaffected():
nonlocal calls
calls += 1
return marker
assert broken.track(unaffected)() is marker and calls == 1
assert isinstance(broken.last_recording_error, MetricsError)
assert broken.last_recording_error.code == "metrics_store_corrupt" and broken.pending_recordings == 0
broken.close()
return {"status": "PASS_INSTALLED_NATIVE_LOCAL_METRICS_COMPONENT", "language": "python", "actualBusinessFileReads": reads,
"persistedOrdinaryObservations": expected, "unscoredObservationsNotCompared": True, "usageAndCostUnmeasured": True,
"originalValueErrorAndCancellationPreserved": True, "recordingFailureDidNotChangeApplication": True,
"csvJsonRoundtrip": True, "reopenedSameNativeStore": True, "benchmark": comparison,
"networkSyscallsAudited": False, "providerUsageExecuted": False, "securityEnforcement": False, "fullTask27Complete": False}Step 10
Run the program with the explicit inputs described above. A missing or invalid input stops the example rather than inventing authority.
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("directory", type=Path)
parser.add_argument("--benchmark-manifest", type=Path)
arguments = parser.parse_args()
print(json.dumps(asyncio.run(main(arguments.directory.resolve(), arguments.benchmark_manifest)), separators=(",", ":")))MCP composition
This complete stdio server calls this feature's real SDK operation. Its bounded tool inputs cannot replace host-selected trust, checkpoint or storage paths. Save it as mcp.mjs and configure your MCP client to launch node mcp.mjs. Public data inputs use the same environment settings as the application program. The code is a custom application tool, not a newly claimed built-in Soterias endpoint.
Install the matching adapter with npm install https://docs.soterias.io/downloads/api-preview/soterias-mcp-server-0.1.0-rc.5-api.2.tgz. Its SDK peer dependency shares the same native class identities as your installed preview.
mcp.mjs
Create mcp.mjs and append these chunks in order. Together they form one complete program; indentation carries across chunks.
Step 1
Import the SDK and standard-library helpers. These are the real package exports used by the executable program.
import { Soterias, observe, canonicalize, Metrics, audit, HistoricalPolicyInputs } from '@soterias/core';
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { McpServer } from '@modelcontextprotocol/server';
import { serveStdio } from '@modelcontextprotocol/server/stdio';
import * as z from 'zod/v4';
// Trusted startup: paths, trust and credentials are fixed by the host.
const metrics = new Metrics({workflow: 'accounting', directory: process.env.SOTERIAS_METRICS});
// Agent input is bounded data. It cannot replace host configuration or trust.
const handle = serveStdio(() => {
const server = new McpServer({name: 'soterias-comparisons', version: '1.0.0'});
server.registerTool('compare_runs', {
description: 'Use comparisons through the host-selected native SDK.',
inputSchema: z.object({}).strict(),
}, async input => {
try {
const result = await (async () => {
return metrics.compare(process.env.SOTERIAS_COHORT, 'baseline', 'instrumented').data();
})();
return {content: [{type: 'text', text: JSON.stringify(result, (_key, value) => typeof value === 'bigint' ? value.toString() : value)}]};
} catch {Step 2
Finish outstanding work and release owned resources. Cleanup errors must not silently replace an existing application error.
return {isError: true, content: [{type: 'text', text: 'feature_request_refused'}]};
}
});
return server;
}, {maxSubscriptions: 0});
process.stdin.once('end', () => {
void (async () => { await handle.close(); await metrics.closeAsync(); })();
});Test the tool over a real MCP connection
Save this client next to mcp.mjs, then run node client.mjs. It starts the host, calls the actual feature tool and closes the connection. It does more than list the tool catalog.
client.mjs
Create client.mjs and append these chunks in order. Together they form one complete program; indentation carries across chunks.
Step 1
Import the SDK and standard-library helpers. These are the real package exports used by the executable program.
import { Client } from '@modelcontextprotocol/client';
import { StdioClientTransport } from '@modelcontextprotocol/client/stdio';
import { fileURLToPath } from 'node:url';
const client = new Client({name: 'feature-example', version: '1.0.0'});
try {
await client.connect(new StdioClientTransport({
command: process.execPath,
args: [fileURLToPath(new URL('./mcp.mjs', import.meta.url))],
env: Object.fromEntries(Object.entries(process.env).filter(([, value]) => value !== undefined)),
}));
const result = await client.callTool({name: 'compare_runs', arguments: {}});
if (result.isError) throw new Error('Feature request refused');
console.log(JSON.stringify(result));
} finally { await client.close(); }Check the result
The application program prints a JSON result and preserves native refusals. For protected operations inspect status and target_observed: transport success does not imply a completed effect. The pure examples contain assertions for their exact expected values. Successful startup is not proof of production enrollment, a physical human review or current authority.
Put it into a workflow
Use the AML walkthrough for protected document reads, reviewed writes and signed outcome checks. Use multi-agent workflows for personal and enterprise orchestration. Keep each agent's proposals separate from operator authority.