SKILL.md

---
name: verifi-harness
description: Author, run and debug a VERIFI engine v2 harness with @pragyacyber/harness-sdk. Use when writing defineHarness, calling ctx.mcp/ctx.emitFinding/ctx.budget, wiring EngineChannel or GrpcEngineChannel, packaging a harness container, or debugging a run that produced no findings.
---

# Authoring a VERIFI harness

## Install, before anything else

The SDKs are PRIVATE packages in GitHub Packages, scope `@pragyacyber`. Not npmjs.

```
# .npmrc — checked in. Token REFERENCED, never written into this file.
@pragyacyber:registry=https://npm.pkg.github.com
//npm.pkg.github.com/:_authToken=${NODE_AUTH_TOKEN}
```

```bash
gh auth refresh -s read:packages          # adds the scope to the gh CLI's token
export NODE_AUTH_TOKEN=$(gh auth token)
npm view @pragyacyber/harness-sdk version # PROVE auth before writing any code
npm i @pragyacyber/harness-sdk@^1.0.0 @pragyacyber/engine-contract@^1.5.0
npm i -D typescript tsx @types/node
```

`404` means the account has no read access to the package — a manual grant by
someone in the org, per user AND per consuming repository. Not fixable from
here: STOP and say so. `401` means the token is missing, unscoped, expired, or
fine-grained (use a classic PAT).

Both packages are `1.x`, so use a CARET. An exact pin was a workaround for `0.x`
caret semantics and is no longer needed. Two copies of `engine-contract` in one
tree break type assignability — check `npm ls @pragyacyber/engine-contract`.

In a Dockerfile the token is a BuildKit secret, NEVER a build ARG — an ARG is
readable in the image history and ships the credential in every layer:

```dockerfile
RUN --mount=type=secret,id=npm_token \
    { echo "@pragyacyber:registry=https://npm.pkg.github.com"; \
      echo "//npm.pkg.github.com/:_authToken=$(cat /run/secrets/npm_token)"; } > .npmrc \
    && npm ci --omit=dev && rm -f .npmrc
```

```bash
DOCKER_BUILDKIT=1 docker build --secret id=npm_token,env=NPM_TOKEN -t h:dev .
```

The harness container serves nothing: no `EXPOSE`, no `HEALTHCHECK`. It dials
out, works, exits.

## Ground yourself first

Before writing any call, open
`node_modules/@pragyacyber/harness-sdk/dist/index.d.ts`. If a symbol is not in
there, it does not exist. Do not infer an API from its name.

Runtime exports, complete: `BudgetLedger`, `ConnectorVersionError`,
`FailoverSlot`, `GrpcEngineChannel`, `MODEL_PRICES_PER_MTOK`, `McpClient`,
`ModelClient`, `ModelRoleNotBoundError`, `SDK_VERSION`, `SlotNotFilledError`,
`collectRunSecrets`, `createMcpResolver`, `createModelResolver`,
`credentialsFromEnv`, `defineHarness`, `i`, `isHarness`, `loadEngineProto`,
`priceFor`, `protoPath`, `redact`, `resolveAwsCredentials`, `runHarness`,
`zPlanDeclaration`.

## The skeleton

```ts
import { defineHarness, i, type HarnessContext } from '@pragyacyber/harness-sdk';

export const harness = defineHarness({
  name: 'my-harness',
  version: '0.1.0',
  connectorVersion: '1.0.0',          // MAJOR must match CONTRACT_VERSION
  inputs: { domain: i.domain({ label: 'Target domain' }) },
  slots: { required: [{ name: 'recon', kind: 'exec' }] },
  models: [{ name: 'analyst', recommendedTier: 'balanced', required: false }],
  phases: [{ key: 'discover', label: 'Discovery', order: 1 }],  // order from 1, unique
  declarePlan: () => ({ phases: [...], coverage: ['...'] }),
  async run(ctx: HarnessContext) {
    return { status: 'done' as const, findings: 0, coverage: [], exitReason: '...' };
  },
});
```

`defineHarness` throws on: blank name; non-semver `version` or
`connectorVersion`; empty `models`; duplicate role names; duplicate required slot
names; duplicate phase `order`s; a phase `order` that is not a positive integer.

## Things that are true and surprising

- `models` cannot be empty even if you never call `ctx.model()`.
- `ctx.mcp(slot).call()` NEVER throws. Check `result.ok`.
- `ctx.mcp(slot).listTools()` DOES throw on a JSON-RPC error.
- `result.ok === true` means the MCP call worked, not that the tool succeeded.
  Parse the tool's own exit status out of the text.
- `ctx.emitFinding` is typed `RawFinding` (from `@pragyacyber/engine-contract`).
  The compiler checks field names now. `cves` is PLURAL; `affected` is
  `{url, parameter}[]`; the stable id is `findingKey`, not `dedupeKey`.
- `ctx.input` is `manifest.parameters`. Your SCOPE is `ctx.target.scope`, which
  IS on the context, along with `ctx.hardDeadline`. Check what you touch against
  it rather than trusting an input parameter.
- `ctx.model()` throws only when NO AWS credentials resolve at all (normal
  locally; on Fargate it reads the task-role endpoint, not env vars). It builds
  every request through the contract, so every modelled provider works — `bedrock`
  signs with the run's role, keyed providers use the injected key. A failed call
  is `ok: false`, not a throw. (Contract routing: harness-sdk 1.1.0; a keyed role's
  provider/baseUrl/apiKey are read per-role from the injected env since 1.2.0.)
- `ctx.mcp().call()` has a 300s ceiling. `createMcpResolver` accepts a
  `timeoutMs` but `runHarness` does not expose one, so through the real
  entrypoint it stands. Clamp your tool timeouts to 285s server-side.
- Phase `order` must be a POSITIVE INTEGER in both the SDK and the registry.
  Number from 1. `0` and `1.5` throw.
- Redaction is on: `runHarness` calls `collectRunSecrets()` and scrubs THIS
  container's injected secrets from tool output. It never scrubs a credential the
  scan FOUND — that is the finding.

## Emitting results

```ts
ctx.emitFinding({
  findingKey: 'check-id:asset',       // deterministic; no time, no run id, no severity
  title: '...',
  severity: 'medium',                 // critical | high | medium | low | info
  confidence: 'firm',                 // confirmed | firm | tentative
  category: 'http',
  description: '...',                 // real prose, not the title again
  impact: '...',
  remediation: '...',
  cwe: 'CWE-319',                     // string or string[]; the engine keeps the first
  cves: ['CVE-2024-0001'],            // note the PLURAL key; singular `cve` is ignored
  affected: [{ url: 'https://host' }],// an ARRAY OF OBJECTS, never a bare string
});
```

Emit AS YOU PRODUCE. Never batch to the end — the engine persists on arrival, so a
run that dies late keeps everything it already found.

NEVER emit `reviewStatus`, `publishedAt`, `reviewedBy` or `clientVisible` — both
the schema and the engine refuse them.

`info` IS a valid severity. `informational` is REFUSED, and the error names the
value to use. Whether to emit info at all is a judgement call: the reference ASM
harness treats informational output as inventory and does not emit it.

`zRawFinding` is `.passthrough()`, so extra keys survive — attach your own
evidence. They are carried, not normalised.

## Budget

`ctx.budget` meters. It cannot refuse a spend and must never grow the ability to.
Useful reads: `timeRemaining()`, `pastDeadline()`, `overThreshold()`, `spent()`.
`allocate()` returns a sub-ledger — an accounting split, not a cap.

## Running it locally — copy this, it is complete

No engine, no AWS account, no MCP needed. `EngineChannel` is a plain interface;
implementing it is how you develop.

```ts
import type { RunManifest } from '@pragyacyber/engine-contract';
import { runHarness, type EngineChannel } from '@pragyacyber/harness-sdk';
import { harness } from './harness.js';

const channel: EngineChannel = {
  async handshake(cv, name, version) { console.log(`handshake ${name}@${version} (${cv})`); },
  sendEvent: (e) => console.log('event ', e.type, e.payload),
  sendFinding: (f) => console.log('finding', f['title']),
  async sendDone(o) { console.log(`done   ${o.status}: ${o.exitReason}`); },
  onStop: () => {},
  async close() {},
};

const manifest: RunManifest = {
  runId: 'run_local_1',
  correlationId: 'corr_local_1',
  binding: { tenantId: 'local', assessmentId: 'assess_local', assetId: 'asset_local' },
  target: { scope: ['https://example.com'], excludes: [], authMode: 'none' },
  parameters: { domain: 'example.com' },          // add dryRun: true to skip run()
  composition: {
    serviceId: 'svc_local',
    serviceVersion: 1,
    harnessDigest: 'sha256:' + '0'.repeat(64),
    mcpDigests: {}, modelIds: {}, kbContentVersions: {},
    capturedAt: new Date().toISOString(),
  },
  budget: {
    maxDurationMinutes: 15,
    costUsdThresholds: [1, 5, 25],
    tokenThresholds: [100_000, 500_000],
    toolCallThresholds: [50, 200],
    evidenceByteThresholds: [1_000_000, 10_000_000],
    maxSubAgents: 1,
    checkpointEverySeconds: 30,
  },
  policy: { redaction: 'standard', hitl: 'approve_destructive',
            allowedActions: [], deniedActions: [] },
  endpoints: { recon: 'http://127.0.0.1:8000' },  // slot name -> base URL
  credentialRefs: {},
  hardDeadline: new Date(Date.now() + 15 * 60_000).toISOString(),
  contractVersion: '1.5.1',
};

console.log(await runHarness(harness, manifest, channel));
```

```bash
npx tsx local-run.ts
```

`EngineChannel` is exactly those 6 methods. With nothing listening on the
endpoint, the tool call fails, the harness carries on, and the SDK emits
`tool_error` AND `mcp.no_healthy_tool` — you write neither.

Set `parameters.dryRun = true` and `runHarness` never calls `harness.run()`; it
streams the static `declarePlan()` phases and returns.

A green dry run proves WIRING, not findings. Never report it as evidence the
scanner works.

## Verification — required before claiming success

```bash
npx tsc --noEmit
npx tsx local-run.ts     # paste the actual output
```

State explicitly whether a finding was emitted and what events fired. If
`mcp.no_healthy_tool` appeared, the slot had nothing behind it and the run's
thinness is a plumbing failure, not a clean target.

## Registering the image

By DIGEST, never a tag:

```
POST /registries/harnesses
{ "id": "...", "name": "...", "image": "...", "imageDigest": "sha256:<64 hex>" }
```

`imageDigest` must match `/^sha256:[0-9a-f]{64}$/`; a tag is a 400. Without it a
harness registers and sits in `probing` forever. Get the digest from the registry
after a push (`aws ecr describe-images … --query 'imageDetails[0].imageDigest'`),
not from your local tag.

The probe checks the image is scanned, the scan is recent, and nothing is
critical/high. UNSCANNED fails exactly like VULNERABLE.

---
name: verifi-mcp
description: Author and conformance-test a VERIFI MCP tool server in Python with FastMCP. Use when writing @mcp.tool handlers, the /healthz and /test endpoints, debugging a three-gate probe failure, or deciding between sidecar and shared concurrency.
---

# Authoring a VERIFI MCP

Python + FastMCP. `@pragyacyber/mcp-sdk`'s `buildMcpServer` serves a plain-JSON
`{method:'list_tools'}` envelope rather than MCP JSON-RPC, so a server built on it
answers 400 to `initialize` and can never pass gate 3. Its `defineMcp` authoring
rules are right and are the rules below; its server is not a deployment path.

## Setup

Nothing private is involved — an MCP consumes no VERIFI package, so there is no
registry token and no access request. It is a plain Python service.

```bash
python -m venv .venv && . .venv/bin/activate   # Scripts\activate on Windows
pip install "mcp[cli]" anyio starlette uvicorn
```

`FastMCP` here is the one from the OFFICIAL `mcp` Python SDK —
`from mcp.server.fastmcp import FastMCP` — not the standalone `fastmcp`
package. They are different projects with different APIs.

## The contract, in full

```
GET  /healthz  → 2xx. Liveness ONLY. Touches NO backend.
GET  /test     → 2xx AND body {"ok": true, "detail": "..."}. Exercises the REAL backend.
POST /mcp      → real MCP JSON-RPC. tools/list must return >= 1 tool.
```

Serve all three on one base URL with
`mcp.run(transport="streamable-http")`.

## Skeleton

```python
import os
from mcp.server.fastmcp import FastMCP
from starlette.requests import Request
from starlette.responses import JSONResponse
from anyio import to_thread        # to_thread is a SUBMODULE; import it by name

mcp = FastMCP("name", host=os.environ.get("MCP_HOST", "0.0.0.0"),
              port=int(os.environ.get("MCP_PORT", "8000")))

@mcp.tool()
async def do_thing(arg: str, workspace: str = "") -> str:
    """One line saying what this does. Say DESTRUCTIVE here if it is."""
    return await to_thread.run_sync(_blocking, arg, workspace)

@mcp.custom_route("/healthz", methods=["GET"])
async def healthz(request: Request) -> JSONResponse:
    return JSONResponse({"status": "ok"})

@mcp.custom_route("/test", methods=["GET"])
async def test(request: Request) -> JSONResponse:
    ok, detail = _really_check_the_backend()
    return JSONResponse({"ok": ok, "detail": detail}, status_code=200 if ok else 503)

if __name__ == "__main__":
    mcp.run(transport="streamable-http")
```

## Rules

- ZERO TOOLS = gate 3 can never pass. At least one, always.
- BLOCKING SYNC TOOL = the event loop freezes for the whole call, so `/healthz`
  and `/test` stop answering and the registry marks you unavailable DURING every
  scan. Always `async def` + `to_thread.run_sync`.
- `/healthz` must not touch the backend. An orchestrator kills containers on it.
- `/test` must EXECUTE its dependencies, not just `shutil.which` them. A
  truncated download resolves on PATH and fails on every real call.
- Split required vs optional dependencies; fail gate 2 only on required.
- Derive every path from a caller-supplied `workspace`. Guard `..` and absolute
  paths. Never fall back to `$HOME`.
- Set `TMPDIR` and `HOME` per call, or concurrent runs share tool caches.
- Clamp the caller's timeout to a server ceiling. Use `start_new_session=True` so
  a timeout kills the process group.
- Truncate large output and SAY you truncated.
- Report every failure in the return value. Never return an empty success.
- Do not accept credentials as tool arguments. Read them from the environment.
  But never filter a credential your tool FOUND on the target out of its own
  output — that is the finding.
- `shared` is not representable on a registration; the enum excludes it. It
  requires a PASSED interleaved-session statelessness test. Register as
  `sidecar` (the default) and say how you tested it when you ask for the
  widening.
- A registration cannot carry `status`, `probe` or `tools` — the schemas are
  `.strict()` and those are outcomes. One of `endpoint` or `image` is required.

## Verification — required before claiming success

```bash
BASE=http://127.0.0.1:8000
curl -sS -o /dev/null -w 'healthz: %{http_code}\n' "$BASE/healthz"
curl -sS -w '\ntest: %{http_code}\n' "$BASE/test"

ACCEPT='Accept: application/json, text/event-stream'
CT='Content-Type: application/json'
SID=$(curl -sS -D - -o /dev/null -X POST "$BASE/mcp" -H "$CT" -H "$ACCEPT" \
  -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"probe","version":"1.0.0"}}}' \
  | tr -d '\r' | awk -F': ' '/^mcp-session-id:/ {print $2}')
curl -sS -o /dev/null -X POST "$BASE/mcp" -H "$CT" -H "$ACCEPT" -H "mcp-session-id: $SID" \
  -d '{"jsonrpc":"2.0","method":"notifications/initialized","params":{}}'
curl -sS -X POST "$BASE/mcp" -H "$CT" -H "$ACCEPT" -H "mcp-session-id: $SID" \
  -d '{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}'
```

All three must pass, and you must COUNT the tools yourself. Also run the probe
while the server is under load — a server that passes at rest and fails during a
scan is worse than one that fails outright.

Paste the real output. Never claim a gate passed without showing the response.

## Container

```dockerfile
FROM python:3.12-slim
WORKDIR /app
RUN pip install --no-cache-dir "mcp[cli]" anyio starlette uvicorn
COPY server.py ./
RUN useradd -r -m runner && chown -R runner /app
USER runner
ENV MCP_HOST=0.0.0.0 MCP_PORT=8000
EXPOSE 8000
HEALTHCHECK CMD python -c "import urllib.request;urllib.request.urlopen('http://127.0.0.1:8000/healthz')"
CMD ["python", "server.py"]
```

Unlike a harness, an MCP DOES serve — it has an `EXPOSE` and a `HEALTHCHECK`,
and the healthcheck points at `/healthz` (liveness) and never at `/test`. An
orchestrator kills containers on a failed healthcheck, and `/test` failing means
a broken backend, not a dead process.