Building an MCP

An MCP is a tool server. A harness reaches it through a slot, and the engine will not bind a slot to it until it has passed a three-gate probe.

What the engine actually requires

The contract is HTTP + MCP streamable-http. Nothing about it is language-specific.

That is the whole contract. Everything else on these pages is either how to satisfy it well, or a rule that stops you satisfying it dishonestly.

The smallest registerable MCP

"""A minimal VERIFI-conformant MCP."""

import os

from mcp.server.fastmcp import FastMCP
from starlette.requests import Request
from starlette.responses import JSONResponse

SERVICE = "example-mcp"
VERSION = "1.0.0"

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


@mcp.tool()
def reverse_dns(ip: str) -> str:
    """Resolve an IP address to its PTR record."""
    import socket

    try:
        return socket.gethostbyaddr(ip)[0]
    except OSError as e:
        return f"[error: {type(e).__name__}: {e}]"


@mcp.custom_route("/healthz", methods=["GET"])
async def healthz(request: Request) -> JSONResponse:
    """Liveness ONLY. Deliberately touches no backend."""
    return JSONResponse({"status": "ok", "service": SERVICE, "version": VERSION})


@mcp.custom_route("/test", methods=["GET"])
async def test(request: Request) -> JSONResponse:
    """Readiness — must exercise the real backend."""
    import socket

    try:
        socket.gethostbyname("localhost")
        ok, detail = True, "resolver reachable"
    except OSError as e:
        ok, detail = False, f"resolver unusable: {e}"
    return JSONResponse(
        {"ok": ok, "detail": detail, "service": SERVICE, "version": VERSION},
        status_code=200 if ok else 503,
    )


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

mcp>=1.2.0
starlette
uvicorn

mcp.run(transport="streamable-http") serves the MCP protocol at /mcp and your custom_route handlers alongside it, on the same port. That is exactly what the probe expects: one base URL, three paths.

The TypeScript MCP SDK

@pragyacyber/mcp-sdk @ 0.1.0 exports three things:

export function defineMcp(def: McpDefinition): Mcp;          // authoring-time validation
export function isMcp(value: unknown): value is Mcp;
export function buildMcpServer(mcp: Mcp, opts?): FastifyInstance;   // ⚠️ see below
export function probeStatelessness(mcp: Mcp, opts): Promise<ConformanceResult>;

defineMcp is genuinely useful and its rules are the right rules — they are the authoring-time half of everything on these pages, and they are listed in the next section.

Verified by starting one and probing it both ways:

GET  /healthz                                    200  {"ok":true,"name":"demo","version":"1.0.0"}
GET  /test                                       200  {"ok":true,"detail":"backend reachable",…}
POST /mcp  {"method":"list_tools"}               200  {"tools":[{"name":"echo",…}]}
POST /mcp  {"method":"call_tool", …}             200  {"ok":true,"output":"hi"}
POST /mcp  {"jsonrpc":"2.0","method":"initialize",…}   400  {"ok":false,"error":"unknown method 'initialize'"}
POST /mcp  {"jsonrpc":"2.0","method":"tools/list",…}   400  {"ok":false,"error":"unknown method 'tools/list'"}

This is the same mistake the three-gate probe itself once made, in the other direction: an earlier version of gate 3 posted a plausible-looking {"method":"list_tools"} and every real server answered 400. That was caught by running it against a live container. This one has never been run against anything, which is exactly why it survives — a component wired to nothing still passes all of its tests.

So: use defineMcp's rules. Do not deploy buildMcpServer. Listed on Known gaps.

probeStatelessness has a narrower limitation — it takes an in-process TypeScript Mcp and calls tool.handler directly, so it cannot reach a Python server over a socket. Its design is documented on Statelessness and sharing.

Authoring rules the SDK enforces

defineMcp() refuses a definition at authoring time for five reasons. FastMCP cannot enforce them for you, so in Python they become review rules.

test() is a required field on the definition, and its docstring is the rule: it must exercise the real backend. Not "does my process respond" — that is /healthz.

What a slot kind means

kind is a free string naming the capability the server provides — exec, recon, web_proxy, cloud, kb are the ones in use. A harness declares slots: { required: [{ name: 'kali', kind: 'exec' }] }, and an admin binds a server whose kind matches. Nothing in the contract constrains the vocabulary, so match an existing kind rather than inventing one.

Registering one

export const zMcpRegistration = z.object({
  id: z.string().min(1),
  name: z.string().min(1),
  kind: z.string().min(1),
  /** An already-running server to probe. */
  endpoint: z.string().url().optional(),
  /** Or an image to trial: start it, probe it, record what it serves, tear it down. */
  image: z.string().min(1).optional(),
  transport: z.enum(['sse', 'http', 'stdio']).default('http'),
  /** ⛔ `shared` is absent from this enum on purpose. */
  concurrencyModel: zConcurrencyModel.exclude(['shared']).default('sidecar'),
}).strict()
  .refine((m) => Boolean(m.endpoint) || Boolean(m.image), …);

One of endpoint or image is required — with neither there is nothing to probe, and the entry could never leave probing. Verified against contract 1.5.1: either alone is accepted, neither is refused, and the defaults are transport: 'http', concurrencyModel: 'sidecar'.

You cannot send status, probe or tools: those are what a probe found, and the schema is .strict(). You also cannot declare shared — see Statelessness and sharing. Full detail on Registering a component.

Two real MCPs to read

Both are in verifi-mcps/, and both are worth reading in full before you write your own.

asm/server.py — auto-discovers tool modules from a tools/ directory (any module exposing register(mcp) is loaded at startup), and probes a set of required and optional CLI binaries in gate 2.

kali/server.py — a persistent shared execution box. It exposes run_command, end_workspace, put_file, get_file and list_workspace, and it is the sharpest example of the concurrency and isolation rules on these pages.

Isolation on a shared server

If your MCP is shared across runs — and a persistent execution box always is — those runs are scanning different tenants. The Kali MCP takes a workspace argument (the run id) on every tool, executes in that run's private directory with its own TMPDIR and HOME, and refuses any path that escapes it:

_SAFE = re.compile(r"[^a-zA-Z0-9_-]")

def _workspace_dir(workspace: str) -> Path:
    name = _SAFE.sub("_", workspace).strip("_") or "_adhoc"
    d = (RUNS_ROOT / name).resolve()
    if RUNS_ROOT.resolve() not in d.parents and d != RUNS_ROOT.resolve():
        d = RUNS_ROOT / "_adhoc"
    d.mkdir(parents=True, exist_ok=True)
    return d


def _safe_join(workspace: str, relpath: str) -> Optional[Path]:
    """Resolve `relpath` UNDER the run's workspace, or None if it escapes."""
    ws = _workspace_dir(workspace)
    p = (ws / relpath.lstrip("/")).resolve()
    if ws.resolve() != p and ws.resolve() not in p.parents:
        return None
    return p

Note that an unrecognised or empty workspace falls back to a shared _adhoc directory rather than $HOME — a manual probe should not land in another run's directory, and it must never escape the runs root.

Blocking work must not block the loop

This is not theoretical. The Kali MCP's run_command was a blocking def calling subprocess.run, which blocks its thread — and as a sync tool on the event loop it blocked the whole loop for the life of the command, which is minutes to half an hour. Two consequences, both observed:

1. /healthz, /test and tools/list are async def on that same loop, so the engine's three-gate probe timed out for the entire duration of any scan. The registry then marked the box unavailable while it was demonstrably working — and any new run was refused at slot-binding time with slot 'kali' … unavailable.
2. Concurrent tool calls serialised. Five tools fired together all returned at exactly the same 277 seconds, queued behind one another rather than running side by side.

The fix is to offload to a worker thread:

from anyio import to_thread   # `to_thread` is a SUBMODULE — `import anyio` alone
                              # does not expose it on anyio 4.x

@mcp.tool()
async def run_command(command: str, timeout: int = 300, workspace: str = "") -> str:
    """Run a shell command and return combined stdout+stderr."""
    timeout = max(1, min(int(timeout or 300), MAX_TIMEOUT))
    return await to_thread.run_sync(_run_command_blocking, command, timeout, workspace)


def _run_command_blocking(command: str, timeout: int, workspace: str) -> str:
    """The blocking half. Only ever called on a worker thread."""
    ...

Bound everything

The Kali MCP caps the timeout (KALI_MAX_TIMEOUT, default 900s), the output size (KALI_MAX_OUTPUT, default 200 000 chars), and per-file transfers (KALI_MAX_FILE_BYTES, default 30 MB). A caller's requested timeout is clamped to the server's ceiling rather than trusted. On a shared box, one run's stuck tool must never starve the others.

Also give long-running subprocesses their own process group so a timeout kills their children too:

proc = subprocess.run(argv, capture_output=True, text=True, timeout=timeout,
                      cwd=str(cwd), env=env, start_new_session=True)

Report failures; never swallow them

Every failure path in the real MCPs returns a describable string or a structured error. None of them return an empty success.

except subprocess.TimeoutExpired:
    return f"$ {command}\n[timed out after {timeout}s — killed]"
except Exception as e:  # a failed run is reported, never swallowed
    return f"$ {command}\n[error: {type(e).__name__}: {e}]"

A swallowed exception becomes "no result" becomes "fine" — and "fine" with zero findings reads as a clean bill of health.