---
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

Full reference: https://docs.verifi.pragyacyber.com/

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.
