A command-line tool you already trust makes a good MCP tool: it has a stable interface, it is installed on the box, and its behavior is documented. The work is not reimplementing it, it is exposing it safely — mapping subcommands and flags to typed inputs, feeding untrusted input as data rather than shell, bounding runtime, and turning a non-zero exit into a clean tool error instead of a stack trace.

This tutorial wraps jq with FastMCP. jq is a good example because it is pure and deterministic: give it a filter and a JSON document and you get the same output every time, with meaningful exit codes when a filter or the input is bad. You will build one hardened subprocess wrapper, expose a general tool and a deliberately constrained one on top of it, and test both the mapping and the safety rules. The stack is Mac-native: uv, make, and pytest.

The safety rules for wrapping a CLI

Every one of these is a line in the wrapper you are about to write. They apply to any binary, not just jq:

  • Never build a shell string. Pass an argument list to subprocess.run and leave shell=False (the default). A filter or payload can then never inject ;, |, $(...), or backticks.
  • Untrusted input goes on stdin, not the command line. Large or hostile JSON is data piped to the process, never an argument the shell or the tool has to parse as a flag.
  • Bound every run with a timeout. A pathological input must not hang the server.
  • Cap the captured output. A tool result that returns megabytes can break the transport; refuse it instead.
  • Normalize failures. Missing binary, timeout, and non-zero exit all become one error type the server converts into an MCP tool error carrying the tool’s own stderr.

What you will build

  • runner.py: the single module that touches subprocess, enforcing all five rules above.
  • server.py: two tools — jq_filter (arbitrary program plus output flags) and jq_keys (a constrained wrapper that can only list keys) — plus a jq://version resource.
  • client.py: an in-memory client that calls each tool, including a failing one.
  • A pytest suite covering the flag mapping, the constrained tool, shell-injection resistance, and the timeout.
  • A make wrapper with a help screen.

Prerequisites

  • macOS 13+ with Homebrew (brew.sh).
  • jq 1.7+ — brew install jq; verify jq --version.
  • uv 0.5+ — brew install uv; verify uv --version.
  • Xcode Command Line Tools (xcode-select --install) for make.
  • Familiarity with FastMCP tools and the in-memory Client (see Build an MCP Server with FastMCP and Design Great MCP Tools: Annotations and Semantics).

Step 1: Add project hygiene

Create the file

mkdir -p wrap-local-cli-mcp-macos
cd wrap-local-cli-mcp-macos
touch .gitignore

Add the code: .gitignore

# Python
__pycache__/
*.py[cod]
.venv/
.uv/
.pytest_cache/
.ruff_cache/
.mypy_cache/
*.log
# OS / editor noise
.DS_Store

Detailed breakdown

  • Standard Python ignores plus .DS_Store, created first so the virtualenv and caches never get committed.

Step 2: Initialize the project with uv

Create the files

uv init --name macmcp --no-workspace
rm -f main.py hello.py
uv add fastmcp
uv add --dev pytest pytest-asyncio

Add the code: pyproject.toml

[project]
name = "macmcp"
version = "0.1.0"
description = "Wrap the jq command-line tool as MCP tools with FastMCP"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "fastmcp>=3.4.4",
]

[dependency-groups]
dev = [
    "pytest>=9.1.1",
    "pytest-asyncio>=1.4.0",
]

Detailed breakdown

  • The only runtime dependency is FastMCP. The tool being wrapped is an external binary on PATH, not a Python package, so it is a system prerequisite rather than a dependency.

Step 3: The hardened subprocess wrapper

Put every subprocess concern in one place. Nothing else in the project calls subprocess, so all the safety rules live and are tested here.

Create the file

touch runner.py

Add the code: runner.py

"""Run the `jq` CLI safely from Python.

The whole point of wrapping a command-line tool for MCP is to invoke it the way a
careful operator would, not the way a shell one-liner does. This module is the one
place that touches `subprocess`, and it enforces four rules every wrapped CLI
should follow:

1. Build an **argument list**, never a shell string — no `shell=True`, so a filter
   or JSON payload can never inject `;`, `|`, `$(...)`, or backticks.
2. Feed untrusted input over **stdin**, not as an argument, so large or hostile
   payloads are data, not command line.
3. Bound every run with a **timeout** so a pathological input cannot hang the
   server.
4. **Cap the captured output** so a tool result cannot blow up the transport.

Errors from the CLI (missing binary, timeout, non-zero exit) are normalized into a
single `CLIError` the server layer turns into a clean MCP tool error.
"""

from __future__ import annotations

import shutil
import subprocess

# The binary we wrap. Resolved once so a clear error fires if it is not installed.
JQ = "jq"

DEFAULT_TIMEOUT = 5.0
MAX_OUTPUT_BYTES = 256 * 1024  # 256 KiB is plenty for a tool result.


class CLIError(Exception):
    """A wrapped-CLI invocation failed in a way the caller should see."""


def run_jq(
    filter_expr: str,
    json_input: str,
    *,
    raw: bool = False,
    compact: bool = False,
    sort_keys: bool = False,
    timeout: float = DEFAULT_TIMEOUT,
) -> str:
    """Run `jq [flags] <filter>` with `json_input` on stdin and return stdout.

    Flags are typed booleans that map to jq options, so the caller never assembles
    a command line: `raw` -> `-r`, `compact` -> `-c`, `sort_keys` -> `-S`.
    """
    if shutil.which(JQ) is None:
        raise CLIError(f"'{JQ}' is not installed or not on PATH")

    args = [JQ]
    if raw:
        args.append("-r")
    if compact:
        args.append("-c")
    if sort_keys:
        args.append("-S")
    args.append(filter_expr)  # a normal argv entry, not shell-interpolated

    try:
        proc = subprocess.run(
            args,
            input=json_input,
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False,
        )
    except subprocess.TimeoutExpired as exc:
        raise CLIError(f"jq timed out after {timeout:g}s") from exc

    if proc.returncode != 0:
        # jq writes a human-readable reason to stderr; surface it verbatim.
        reason = proc.stderr.strip() or f"exit code {proc.returncode}"
        raise CLIError(f"jq failed: {reason}")

    out = proc.stdout
    if len(out.encode()) > MAX_OUTPUT_BYTES:
        raise CLIError(f"jq output exceeded {MAX_OUTPUT_BYTES} bytes")
    return out


def jq_version_string(timeout: float = 2.0) -> str:
    """Return the installed jq version, e.g. `jq-1.8.2`."""
    if shutil.which(JQ) is None:
        raise CLIError(f"'{JQ}' is not installed or not on PATH")
    try:
        proc = subprocess.run(
            [JQ, "--version"],
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False,
        )
    except subprocess.TimeoutExpired as exc:
        raise CLIError(f"jq timed out after {timeout:g}s") from exc
    return proc.stdout.strip()

Detailed breakdown

  • args = [JQ] then args.append(...) builds a real argument vector. Because subprocess.run gets a list and shell stays False, the filter is handed to jq as one argument via execvp; there is no shell to interpret metacharacters. A filter of . ; rm -rf ~ is just an invalid jq program, not a second command.
  • input=json_input streams the JSON to jq’s stdin. The document never appears on the command line, so its size and contents cannot overflow argv or be misread as a flag.
  • timeout=timeout turns a runaway program (jq can loop forever with something like while(true; .)) into a TimeoutExpired, which becomes a CLIError.
  • The returncode check captures jq’s own stderr message so the caller learns why the filter failed, not just that it did.
  • len(out.encode()) > MAX_OUTPUT_BYTES rejects an oversized result rather than shipping it through the transport.
  • shutil.which(JQ) gives a readable “not installed” error instead of a bare FileNotFoundError if jq is missing.

Step 4: Expose a general tool and a constrained one

Two tools wrap the same binary. jq_filter is the full-power operation; jq_keys is deliberately narrow — it runs a fixed program and takes no filter at all. Giving a model the smallest capability that does the job is the safe default.

Create the file

touch server.py

Add the code: server.py

"""Expose the `jq` CLI as MCP tools.

Two tools, both read-only, wrap one binary:

- `jq_filter` is the general operation — an arbitrary jq program plus the common
  output flags, mapped to typed inputs.
- `jq_keys` is a *constrained* operation — it runs the fixed `keys` program, so a
  caller that only needs object keys cannot pass a free-form filter at all.

Exposing a narrow tool alongside the general one is the safe-by-default pattern:
give a model the smallest capability that does the job. A `jq://version` resource
reports which jq is installed.

Run over stdio with `python server.py`, or drive it with client.py.
"""

from fastmcp import FastMCP
from fastmcp.exceptions import ToolError
from mcp.types import ToolAnnotations

from runner import CLIError, jq_version_string, run_jq

mcp = FastMCP(name="jq-tools")

READ_ONLY = ToolAnnotations(readOnlyHint=True, idempotentHint=True, openWorldHint=False)


@mcp.tool(annotations=READ_ONLY)
def jq_filter(
    filter_expr: str,
    json_input: str,
    raw: bool = False,
    compact: bool = False,
    sort_keys: bool = False,
) -> str:
    """Run a jq program over a JSON document.

    Args:
        filter_expr: a jq filter, e.g. ".users[].name" or "keys".
        json_input: the JSON document to read (passed to jq on stdin).
        raw: emit raw strings without JSON quotes (jq -r).
        compact: emit compact single-line JSON (jq -c).
        sort_keys: sort object keys in the output (jq -S).
    """
    try:
        return run_jq(
            filter_expr,
            json_input,
            raw=raw,
            compact=compact,
            sort_keys=sort_keys,
        )
    except CLIError as exc:
        raise ToolError(str(exc)) from exc


@mcp.tool(annotations=READ_ONLY)
def jq_keys(json_input: str) -> str:
    """List the keys of a JSON object, sorted, as a compact array.

    A constrained wrapper: it always runs the `keys` program, so no arbitrary jq
    filter can be supplied through this tool.
    """
    try:
        return run_jq("keys", json_input, compact=True)
    except CLIError as exc:
        raise ToolError(str(exc)) from exc


@mcp.resource("jq://version")
def jq_version() -> str:
    """The version of the jq binary this server shells out to."""
    try:
        return jq_version_string()
    except CLIError as exc:
        raise ToolError(str(exc)) from exc


if __name__ == "__main__":
    mcp.run()

Detailed breakdown

  • jq_filter’s signature is the flag map. Each boolean parameter corresponds to one jq option; FastMCP builds the tool’s input schema from the annotations, so the model sees typed raw/compact/sort_keys switches instead of a raw command line. The docstring Args: become the parameter descriptions.
  • jq_keys takes only json_input. It hardcodes the keys program, so this tool cannot run an arbitrary filter no matter what a caller sends. When you only need one operation, expose that operation, not the whole tool.
  • ToolAnnotations(readOnlyHint=True, ...) advertises that jq neither mutates state nor reaches the network. These are advisory hints surfaced in tools/list, covered in the annotations article.
  • except CLIError -> raise ToolError is the boundary: the runner raises a plain CLIError, and the server converts it into a ToolError, which the client receives as a readable tool error carrying jq’s stderr.
  • The jq://version resource reports the installed jq. Its value depends on the machine, so treat it as informational rather than a fixed string.

Step 5: Drive it in-memory

Create the file

touch client.py

Add the code: client.py

"""Drive the jq-tools server in-memory.

Lists the two tools and the version resource, then calls each tool: a general
filter (compact output), the constrained keys tool, and a filter that fails so you
can see the CLI's error surfaced as a clean MCP error.
"""

import asyncio

from fastmcp import Client
from fastmcp.exceptions import ToolError

from server import mcp

SAMPLE = '{"name": "Ada", "langs": ["ada", "python"], "active": true}'


async def main() -> None:
    async with Client(mcp) as client:
        print("tools:    ", [t.name for t in await client.list_tools()])
        print("resources:", [str(r.uri) for r in await client.list_resources()])
        print("version:  ", (await client.read_resource("jq://version"))[0].text)

        print("\njq_filter '.langs' (compact):")
        r = await client.call_tool(
            "jq_filter", {"filter_expr": ".langs", "json_input": SAMPLE, "compact": True}
        )
        print("  ", r.data.rstrip())

        print("jq_filter '.name' (raw):")
        r = await client.call_tool(
            "jq_filter", {"filter_expr": ".name", "json_input": SAMPLE, "raw": True}
        )
        print("  ", r.data.rstrip())

        print("jq_keys:")
        r = await client.call_tool("jq_keys", {"json_input": SAMPLE})
        print("  ", r.data.rstrip())

        print("\njq_filter with a bad filter (errors):")
        try:
            await client.call_tool(
                "jq_filter", {"filter_expr": ".[]", "json_input": "5"}
            )
        except ToolError as exc:
            print("  ToolError:", exc)


if __name__ == "__main__":
    asyncio.run(main())

Detailed breakdown

  • r.data is the tool’s string return value (jq’s stdout, which ends in a newline, so the demo .rstrip()s it for display).
  • The .langs and .name calls show the flag mapping in action: compact=True yields single-line JSON, raw=True drops the quotes around a string.
  • jq_keys returns the sorted key array with no filter argument supplied.
  • The failing call passes .[] against the number 5, which jq rejects; the CLIError becomes a ToolError the client catches.

Run it:

uv run python client.py

Expected output (jq’s version reflects your install):

tools:     ['jq_filter', 'jq_keys']
resources: ['jq://version']
version:   jq-1.8.2

jq_filter '.langs' (compact):
   ["ada","python"]
jq_filter '.name' (raw):
   Ada
jq_keys:
   ["active","langs","name"]

jq_filter with a bad filter (errors):
  ToolError: jq failed: jq: error (at <stdin>:0): Cannot iterate over number (5)

FastMCP also logs the caught tool error to stderr (a line like Error calling tool 'jq_filter'); that is expected and separate from the stdout above.

Step 6: Test the mapping and the safety rules

Create the files

mkdir -p tests
touch tests/__init__.py
touch pytest.ini
touch tests/test_server.py

Add the code: pytest.ini

[pytest]
asyncio_mode = auto
filterwarnings =
    ignore::DeprecationWarning

Detailed breakdown

  • asyncio_mode = auto runs the async tests without a marker.
  • tests/__init__.py makes tests/ a package so the project root lands on sys.path and from server import mcp resolves.

Add the code: tests/test_server.py

"""Tests for the jq-wrapping MCP server.

Two layers:

- the `runner` module, where the subprocess safety rules live (no shell,
  timeouts, error normalization);
- the MCP server, where the two tools and the version resource are exposed.
"""

import pytest
from fastmcp import Client
from fastmcp.exceptions import ToolError

from runner import CLIError, run_jq
from server import mcp

SAMPLE = '{"name": "Ada", "langs": ["ada", "python"], "active": true}'


# --- runner-level tests -------------------------------------------------------


def test_compact_and_raw_flags_map_to_options():
    assert run_jq(".langs", SAMPLE, compact=True) == '["ada","python"]\n'
    assert run_jq(".name", SAMPLE, raw=True) == "Ada\n"


def test_sort_keys_flag():
    assert run_jq(".", '{"b":2,"a":1}', compact=True, sort_keys=True) == '{"a":1,"b":2}\n'


def test_nonzero_exit_becomes_clierror():
    with pytest.raises(CLIError) as exc:
        run_jq(".[]", "5")  # cannot iterate over a number
    assert "jq failed" in str(exc.value)


def test_filter_metacharacters_are_not_shell_interpreted(tmp_path):
    """A filter containing shell syntax is passed to jq as data, not a command."""
    marker = tmp_path / "pwned"
    with pytest.raises(CLIError):
        run_jq(f". ; touch {marker}", "{}")
    assert not marker.exists()


def test_timeout_is_enforced():
    with pytest.raises(CLIError) as exc:
        run_jq("while(true; .)", "0", timeout=0.5)
    assert "timed out" in str(exc.value)


# --- server-level tests -------------------------------------------------------


async def test_tools_are_listed_and_read_only():
    async with Client(mcp) as client:
        tools = {t.name: t for t in await client.list_tools()}
        assert set(tools) == {"jq_filter", "jq_keys"}
        assert tools["jq_filter"].annotations.readOnlyHint is True


async def test_jq_filter_compact():
    async with Client(mcp) as client:
        r = await client.call_tool(
            "jq_filter", {"filter_expr": ".langs", "json_input": SAMPLE, "compact": True}
        )
        assert r.data == '["ada","python"]\n'


async def test_jq_keys_is_sorted_and_compact():
    async with Client(mcp) as client:
        r = await client.call_tool("jq_keys", {"json_input": SAMPLE})
        assert r.data == '["active","langs","name"]\n'


async def test_jq_keys_has_no_filter_argument():
    """The constrained tool exposes only json_input, never a free-form filter."""
    async with Client(mcp) as client:
        tools = {t.name: t for t in await client.list_tools()}
        assert set(tools["jq_keys"].inputSchema["properties"]) == {"json_input"}


async def test_bad_filter_surfaces_as_tool_error():
    async with Client(mcp) as client:
        with pytest.raises(ToolError):
            await client.call_tool(
                "jq_filter", {"filter_expr": ".[]", "json_input": "5"}
            )


async def test_version_resource_reports_jq():
    async with Client(mcp) as client:
        text = (await client.read_resource("jq://version"))[0].text
        assert text.startswith("jq-")

Detailed breakdown

  • test_filter_metacharacters_are_not_shell_interpreted is the security test. It passes a filter that would create a file if a shell ran it, then asserts the file does not exist — proving the argument list, not a shell, executed jq.
  • test_timeout_is_enforced feeds jq an infinite program and a short timeout, confirming the runner cuts it off instead of hanging.
  • test_jq_keys_has_no_filter_argument pins the constraint: the narrow tool’s input schema has exactly one property, so no arbitrary filter can reach jq through it.
  • The runner tests call run_jq directly; the server tests go through the MCP Client, so both the subprocess layer and the tool layer are covered.

Step 7: Wrap it in a Makefile

Create the file

touch Makefile

Add the code: Makefile

.DEFAULT_GOAL := help

.PHONY: help install demo serve test clean

help:  ## Show this help screen
	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) \
		| awk 'BEGIN {FS = ":.*?## "}; {printf "  \033[36m%-10s\033[0m %s\n", $$1, $$2}'

install:  ## Sync runtime and dev dependencies
	uv sync

demo:  ## Wrap the jq CLI and call each tool
	uv run python client.py

serve:  ## Run the jq-tools server over stdio (for a real MCP client)
	uv run python server.py

test:  ## Run the test suite
	uv run pytest -q

clean:  ## Remove caches
	rm -rf .pytest_cache __pycache__ tests/__pycache__

Detailed breakdown

  • .DEFAULT_GOAL := help makes a bare make print the help screen. Recipe bodies must be tab-indented or make errors.

Step 8: Run everything

make          # prints the help screen
make demo     # wraps jq and calls each tool (output in Step 5)
make test     # runs the suite

The suite passes:

...........                                                              [100%]
11 passed in 0.83s

Notes on wrapping other CLIs

  • Subcommands map to tools. jq is flag-driven, so its options became typed parameters. For a subcommand-style tool (git log, docker ps, kubectl get), give each subcommand its own tool with its own typed inputs, exactly the way jq_filter and jq_keys are separate tools here.
  • Prefer an allow-list. Do not expose a generic “run any argument” tool. Model the specific operations you want, and constrain their inputs (as jq_keys does).
  • Keep the subprocess in one module. One runner.py that every tool goes through means the no-shell, timeout, and output-cap rules are written and tested once.
  • Resolve the binary explicitly. shutil.which up front turns a missing tool into a clear message. For extra safety, pin an absolute path so PATH changes cannot swap the binary.
  • Return exit codes as errors, not silence. A non-zero exit should raise, with the tool’s stderr attached, so the model can correct its input.

Troubleshooting

  • 'jq' is not installed or not on PATH. Install it with brew install jq, or point JQ at an absolute path. A GUI-launched MCP host may have a narrower PATH than your shell.
  • A filter with a ; or | “does nothing” / errors as a syntax error. That is correct — it reached jq as a filter, not a shell. There is no shell to run it.
  • A tool call hangs. Some jq programs loop forever; the timeout bounds them. Lower DEFAULT_TIMEOUT if your inputs should always be fast.
  • AttributeError on r.data. Older FastMCP exposes the string on r.content[0].text; r.data holds the typed return value on current versions.
  • ModuleNotFoundError: No module named 'server' under pytest. tests/ needs an __init__.py so pytest puts the project root on sys.path.

Recap

One binary became two MCP tools with no reimplementation: jq_filter maps jq’s flags to typed parameters, and jq_keys exposes a single constrained operation. A single runner.py holds every subprocess safety rule — argument lists instead of a shell, JSON on stdin, a timeout, an output cap, and normalized errors — and the test suite proves the important ones, including that shell metacharacters in a filter never execute. The same shape wraps any CLI you already trust.

Next improvements:

  • Register the server over stdio with make serve and add it to Claude Desktop or Claude Code.
  • Wrap a subcommand-style tool (git, docker) with one tool per subcommand.
  • Add a per-call byte budget on stdin as well as stdout for tools that accept large documents.