Generate Synthetic JSON Requests to Test an API on macOS built two generators and a CLI that fires batches at an order-intake API. This article puts the same generators behind an MCP server, so an agent can preview a payload, check the target, and trigger a batch by asking for one.

Wrapping a generator in tools is the easy half. The half worth attention is the control surface: which decisions the caller gets to make and which ones the server keeps. A model that can pick the destination of a traffic generator is a server-side request forgery primitive with a friendly name, and a model that can pick the batch size can turn one sentence into fifty thousand POSTs. Here the target comes from the environment and the batch size is capped, so the tools stay useful without handing over either decision.

The project is self-contained. Files carried over from the previous article are reproduced in full and marked as unchanged, so you can follow this one on its own.

Prerequisites

Step 1: Scaffold the project

Create the file

uv init synthetic-orders-mcp
cd synthetic-orders-mcp
rm main.py README.md
uv add fastmcp fastapi "uvicorn[standard]" httpx
uv add --dev pytest pytest-asyncio
mkdir -p api synth data tests scripts
touch api/__init__.py synth/__init__.py
touch .gitignore

Add the code: .gitignore

__pycache__/
*.pyc
.venv/
.pytest_cache/
*.egg-info/
dist/
build/
.DS_Store
*.log
tmp/
.mcp.json

Detailed breakdown

  • fastmcp provides the server. fastapi and uvicorn[standard] run the order API the server sends traffic to, and httpx is how the server reaches it.
  • pytest-asyncio is needed because the MCP client API is async; the asyncio_mode = "auto" setting added in Step 10 keeps the tests free of per-test decorators.
  • .mcp.json is ignored on purpose. It carries an absolute path that is correct only on the machine that generated it, so it is rendered by make config from a committed template rather than checked in. The template itself is tracked.
  • Everything else is the standard Python artifact list from the previous article.

Step 2: Bring over the seed database

Unchanged from the previous article. The API validates against these rows, and the generator samples from them, which is what keeps generated traffic consistent with what the API will accept.

Create the file

touch data/catalog.json
touch data/customers.json

Add the code: data/catalog.json

[
  { "sku": "SKU-4021", "name": "Trail Runner 12", "category": "footwear", "price_cents": 12900, "popularity": 34, "in_stock": true },
  { "sku": "SKU-4022", "name": "Trail Runner 12 Wide", "category": "footwear", "price_cents": 12900, "popularity": 9, "in_stock": true },
  { "sku": "SKU-5510", "name": "Merino Crew Sock", "category": "apparel", "price_cents": 1850, "popularity": 61, "in_stock": true },
  { "sku": "SKU-5511", "name": "Merino Base Layer", "category": "apparel", "price_cents": 7400, "popularity": 22, "in_stock": true },
  { "sku": "SKU-6300", "name": "Alloy Water Bottle", "category": "gear", "price_cents": 2450, "popularity": 47, "in_stock": true },
  { "sku": "SKU-6301", "name": "Insulated Flask 1L", "category": "gear", "price_cents": 3900, "popularity": 15, "in_stock": true },
  { "sku": "SKU-7100", "name": "Headlamp 400", "category": "gear", "price_cents": 5600, "popularity": 28, "in_stock": true },
  { "sku": "SKU-7101", "name": "Headlamp 400 Rechargeable", "category": "gear", "price_cents": 8200, "popularity": 12, "in_stock": false }
]

Detailed breakdown

  • sku is the join key between the generator and the API, and price_cents is an integer so totals computed on both sides agree exactly.
  • popularity is a relative weight the generator uses to pick common products more often than rare ones.
  • SKU-7101 is out of stock, so the generator must skip it and the API must reject it. It is also what the catalog resource in Step 6 is checked against.

Add the code: data/customers.json

[
  { "customer_id": "CUST-0001", "name": "Ada Rowan", "tier": "gold", "channels": ["web", "mobile"] },
  { "customer_id": "CUST-0002", "name": "Bo Whitfield", "tier": "standard", "channels": ["web"] },
  { "customer_id": "CUST-0003", "name": "Cleo Nakamura", "tier": "standard", "channels": ["mobile"] },
  { "customer_id": "CUST-0004", "name": "Dev Okonkwo", "tier": "gold", "channels": ["web", "partner"] },
  { "customer_id": "CUST-0005", "name": "Elin Vasquez", "tier": "partner", "channels": ["partner"] }
]

Detailed breakdown

  • channels is a per-customer allowlist, so a valid request needs a channel that is legal for that customer. The seeded generator reads the constraint out of the data rather than restating it.
  • The CUST-#### identifier format matters later: the naive generator invents C-#### instead, which is why its requests are rejected every time.

Step 3: Bring over the order API

Unchanged from the previous article. It validates in two layers, and that split is what makes the two generator modes behave differently.

Create the file

touch api/main.py

Add the code: api/main.py

"""Order intake API: the target under test for synthetic requests."""

import json
from pathlib import Path
from typing import Literal

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

DATA_DIR = Path(__file__).resolve().parent.parent / "data"


def _load(name: str) -> list[dict]:
    return json.loads((DATA_DIR / name).read_text(encoding="utf-8"))


CATALOG = {row["sku"]: row for row in _load("catalog.json")}
CUSTOMERS = {row["customer_id"]: row for row in _load("customers.json")}


class OrderLine(BaseModel):
    sku: str
    quantity: int = Field(ge=1, le=20)
    unit_price_cents: int = Field(ge=1)


class OrderRequest(BaseModel):
    request_id: str
    customer_id: str
    channel: Literal["web", "mobile", "partner"]
    currency: Literal["USD"]
    lines: list[OrderLine] = Field(min_length=1, max_length=5)
    total_cents: int = Field(ge=1)


app = FastAPI(title="Order Intake")


@app.get("/health")
def health() -> dict:
    return {"status": "ok", "skus": len(CATALOG), "customers": len(CUSTOMERS)}


@app.post("/orders", status_code=201)
def create_order(order: OrderRequest) -> dict:
    errors: list[str] = []

    customer = CUSTOMERS.get(order.customer_id)
    if customer is None:
        errors.append(f"unknown customer {order.customer_id}")
    elif order.channel not in customer["channels"]:
        errors.append(f"channel {order.channel} not enabled for {order.customer_id}")

    expected_total = 0
    for line in order.lines:
        product = CATALOG.get(line.sku)
        if product is None:
            errors.append(f"unknown sku {line.sku}")
            continue
        if not product["in_stock"]:
            errors.append(f"sku {line.sku} is out of stock")
        if product["price_cents"] != line.unit_price_cents:
            errors.append(
                f"price mismatch for {line.sku}: "
                f"sent {line.unit_price_cents}, catalog {product['price_cents']}"
            )
        expected_total += product["price_cents"] * line.quantity

    if not errors and expected_total != order.total_cents:
        errors.append(f"total mismatch: sent {order.total_cents}, expected {expected_total}")

    if errors:
        raise HTTPException(status_code=422, detail=errors)

    return {
        "order_id": f"ORD-{order.request_id[:8]}",
        "customer_id": order.customer_id,
        "lines": len(order.lines),
        "total_cents": order.total_cents,
    }

Detailed breakdown

  • Pydantic enforces structure and the handler enforces referential integrity. The MCP server’s two modes exercise one layer each: seeded traffic clears both, simple traffic clears the first and fails the second.
  • /health returns the row counts, which is what the check_target tool surfaces so an agent can tell “API is down” from “API is up but rejecting everything”.
  • Errors accumulate into a list, so one 422 explains everything wrong with a payload. The MCP layer passes a sample of those strings straight through.
  • The endpoint has no persistence. Accepted orders are validated and echoed, not stored, which is what makes it safe to fire repeated synthetic batches at it.

Step 4: Bring over the generators

Unchanged from the previous article. Three small modules: the seed loader, the naive generator, and the catalog-backed one.

Create the file

touch synth/seed.py
touch synth/simple.py
touch synth/seeded.py

Add the code: synth/seed.py

"""Load the JSON seed database that both the API and the generator read."""

import json
from dataclasses import dataclass
from pathlib import Path

DATA_DIR = Path(__file__).resolve().parent.parent / "data"


@dataclass(frozen=True)
class SeedData:
    products: list[dict]
    customers: list[dict]

    @property
    def sellable(self) -> list[dict]:
        return [p for p in self.products if p["in_stock"]]


def load_seed(data_dir: Path = DATA_DIR) -> SeedData:
    products = json.loads((data_dir / "catalog.json").read_text(encoding="utf-8"))
    customers = json.loads((data_dir / "customers.json").read_text(encoding="utf-8"))
    if not products or not customers:
        raise ValueError(f"seed data under {data_dir} is empty")
    return SeedData(products=products, customers=customers)

Detailed breakdown

  • sellable filters out-of-stock rows at the point of use, so every generator and the catalog resource in Step 6 inherit the rule.
  • load_seed takes a directory argument, which lets a test point at fixture data without touching the real files.
  • The server loads this once at import. A malformed seed file then fails at startup, which an MCP client surfaces as a server that will not connect, rather than as a tool that fails on first call.

Add the code: synth/simple.py

"""Example 1: build request payloads from random primitives only."""

import random
import string
import uuid

CHANNELS = ("web", "mobile", "partner")


def request_id(rng: random.Random) -> str:
    """A seeded stand-in for uuid4() so runs are reproducible."""
    return str(uuid.UUID(int=rng.getrandbits(128), version=4))


def random_sku(rng: random.Random) -> str:
    return "SKU-" + "".join(rng.choices(string.digits, k=4))


def random_line(rng: random.Random) -> dict:
    return {
        "sku": random_sku(rng),
        "quantity": rng.randint(1, 5),
        "unit_price_cents": rng.randrange(500, 15000, 50),
    }


def random_order(rng: random.Random) -> dict:
    lines = [random_line(rng) for _ in range(rng.randint(1, 3))]
    return {
        "request_id": request_id(rng),
        "customer_id": f"C-{rng.randint(1, 9999):04d}",
        "channel": rng.choice(CHANNELS),
        "currency": "USD",
        "lines": lines,
        "total_cents": sum(line["quantity"] * line["unit_price_cents"] for line in lines),
    }

Detailed breakdown

  • Every function takes an rng rather than calling module-level random, which is what makes a batch replayable from its seed. The MCP server depends on this: it reports the seed it used so any run can be repeated exactly.
  • request_id builds a UUID from the seeded generator instead of uuid.uuid4(), which would ignore the seed and break that guarantee.
  • The C-#### customer format is the deliberate flaw. It looks plausible and matches nothing, so this mode is a rejection-path drill.

Add the code: synth/seeded.py

"""Example 2: build request payloads by sampling the JSON seed database."""

import random

from synth.seed import SeedData
from synth.simple import request_id

MAX_LINES = 3
MAX_QUANTITY = 4


def weighted_sample(rng: random.Random, products: list[dict], k: int) -> list[dict]:
    """Pick k distinct products, favoring the ones with higher popularity."""
    pool = list(products)
    picked: list[dict] = []
    for _ in range(min(k, len(pool))):
        weights = [p["popularity"] for p in pool]
        choice = rng.choices(pool, weights=weights, k=1)[0]
        pool.remove(choice)
        picked.append(choice)
    return picked


def random_order(rng: random.Random, seed_data: SeedData) -> dict:
    customer = rng.choice(seed_data.customers)
    products = weighted_sample(rng, seed_data.sellable, rng.randint(1, MAX_LINES))

    lines = [
        {
            "sku": product["sku"],
            "quantity": rng.randint(1, MAX_QUANTITY),
            "unit_price_cents": product["price_cents"],
        }
        for product in products
    ]

    return {
        "request_id": request_id(rng),
        "customer_id": customer["customer_id"],
        "channel": rng.choice(customer["channels"]),
        "currency": "USD",
        "lines": lines,
        "total_cents": sum(line["quantity"] * line["unit_price_cents"] for line in lines),
    }

Detailed breakdown

  • Identities come from the seed database and only the selection is random, which is what makes this mode’s traffic acceptable to the API.
  • weighted_sample removes each pick from a local copy of the pool, so a request never repeats a SKU and the seed rows are left untouched.
  • channel is drawn from the chosen customer’s own allowlist, satisfying the cross-field rule by construction.

Step 5: Put the target and the limits in the environment

This is the first new file, and it is where the security posture of the whole server is decided.

Create the file

touch synth/config.py

Add the code: synth/config.py

"""Server-side settings. Read from the environment, never from a tool argument."""

import os
from dataclasses import dataclass


@dataclass(frozen=True)
class Settings:
    api_url: str
    max_batch: int
    timeout_s: float


def load_settings(env: dict[str, str] | None = None) -> Settings:
    src = os.environ if env is None else env
    max_batch = int(src.get("MAX_BATCH", "100"))
    if max_batch < 1:
        raise ValueError(f"MAX_BATCH must be at least 1, got {max_batch}")
    return Settings(
        api_url=src.get("ORDER_API_URL", "http://127.0.0.1:8000"),
        max_batch=max_batch,
        timeout_s=float(src.get("REQUEST_TIMEOUT_S", "10")),
    )

Detailed breakdown

  • api_url lives here rather than in a tool signature, and that single choice is the article’s main point. Whoever launches the server picks the destination; the model picks how much traffic and of what kind. A target_url parameter would turn this server into a general-purpose HTTP client that a prompt can point anywhere, which is the confused-deputy shape described in Harden an MCP Server: A Threat Model and Defenses on macOS.
  • max_batch is the blast-radius control. Tool arguments come from a model reading text it did not write, so “send ten thousand orders” needs a ceiling that does not depend on the model behaving.
  • Validating MAX_BATCH at load time means a bad value stops the server at startup instead of silently disabling the cap.
  • load_settings accepts an explicit mapping, so tests can build settings without mutating the process environment.
  • The dataclass is frozen, which prevents rebinding these attributes on a shared instance. That is a shallow guarantee, but the values are scalars here, so nothing mutable hides behind it.

Step 6: Write the MCP server

Create the file

touch server.py

Add the code: server.py

"""An MCP server that triggers synthetic order traffic against the order API.

The generators are the ones from the previous article; what is new here is the
control surface. An agent can preview a payload, check the target, and fire a
batch, but it cannot choose *where* the traffic goes or how much of it there is:

- the target URL comes from `ORDER_API_URL` in the server's environment and is
  deliberately not a tool parameter, so a prompt cannot aim the generator at an
  arbitrary host;
- `count` is bounded by `MAX_BATCH`, and exceeding it is a readable error rather
  than a silent clamp;
- every batch reports the seed it used, so any run can be replayed exactly.

Run over stdio with `uv run python server.py`.
"""

import random
from collections import Counter

import httpx
from fastmcp import FastMCP
from fastmcp.exceptions import ToolError
from mcp.types import ToolAnnotations
from pydantic import BaseModel, Field
from typing import Literal

from synth import seeded, simple
from synth.config import load_settings
from synth.seed import load_seed

SETTINGS = load_settings()
SEED_DATA = load_seed()

Mode = Literal["seeded", "simple"]

mcp = FastMCP("synthetic-orders", mask_error_details=True)


class TargetStatus(BaseModel):
    """What the configured order API reports about itself."""

    api_url: str
    reachable: bool
    skus: int | None = None
    customers: int | None = None


class OrderPreview(BaseModel):
    """One generated order, returned without sending it anywhere."""

    mode: Mode
    seed: int
    order: dict
    line_count: int
    total_cents: int


class SendResult(BaseModel):
    """The outcome of one batch, summarized by status code."""

    mode: Mode
    seed: int = Field(description="Replay this batch by passing this seed back in.")
    requested: int
    accepted: int
    rejected: int
    status_counts: dict[str, int]
    accepted_total_cents: int
    sample_failures: list[str]


def make_client() -> httpx.Client:
    """Build the HTTP client used to reach the order API.

    Isolated in a function so tests can swap in an in-process ASGI transport.
    """
    return httpx.Client(base_url=SETTINGS.api_url, timeout=SETTINGS.timeout_s)


def resolve_seed(seed: int | None) -> int:
    """Use the caller's seed, or draw one that the result will report back."""
    if seed is not None:
        return seed
    return random.SystemRandom().randrange(2**31)


def build_batch(mode: Mode, count: int, seed: int) -> list[dict]:
    rng = random.Random(seed)
    if mode == "simple":
        return [simple.random_order(rng) for _ in range(count)]
    return [seeded.random_order(rng, SEED_DATA) for _ in range(count)]


def check_batch_size(count: int) -> None:
    if count < 1:
        raise ToolError(f"count must be at least 1, got {count}")
    if count > SETTINGS.max_batch:
        raise ToolError(
            f"count {count} exceeds the server's MAX_BATCH of {SETTINGS.max_batch}; "
            f"send smaller batches or raise MAX_BATCH in the server environment"
        )


@mcp.resource("synthetic://catalog")
def catalog() -> list[dict]:
    """The sellable catalog rows the seeded generator draws from."""
    return SEED_DATA.sellable


@mcp.tool(
    annotations=ToolAnnotations(
        title="Check the order API",
        readOnlyHint=True,
        idempotentHint=True,
        openWorldHint=True,
    )
)
def check_target() -> TargetStatus:
    """Report whether the configured order API is reachable, and what it holds."""
    try:
        with make_client() as client:
            body = client.get("/health").json()
    except httpx.HTTPError:
        return TargetStatus(api_url=SETTINGS.api_url, reachable=False)
    return TargetStatus(
        api_url=SETTINGS.api_url,
        reachable=True,
        skus=body.get("skus"),
        customers=body.get("customers"),
    )


@mcp.tool(
    annotations=ToolAnnotations(
        title="Preview a synthetic order",
        readOnlyHint=True,
        idempotentHint=True,
        openWorldHint=False,
    )
)
def preview_order(mode: Mode = "seeded", seed: int | None = None) -> OrderPreview:
    """Generate one order and return it without sending it.

    `seeded` samples the catalog and should be accepted; `simple` invents every
    field and should be rejected. Pass `seed` to reproduce an exact payload.
    """
    resolved = resolve_seed(seed)
    order = build_batch(mode, 1, resolved)[0]
    return OrderPreview(
        mode=mode,
        seed=resolved,
        order=order,
        line_count=len(order["lines"]),
        total_cents=order["total_cents"],
    )


@mcp.tool(
    annotations=ToolAnnotations(
        title="Send synthetic orders",
        readOnlyHint=False,
        destructiveHint=False,
        idempotentHint=False,
        openWorldHint=True,
    )
)
def send_orders(count: int = 10, mode: Mode = "seeded", seed: int | None = None) -> SendResult:
    """Generate `count` orders and POST each one to the configured order API.

    Use `seeded` for traffic that should be accepted and `simple` to drill the
    API's rejection path. The result reports the seed, so the same batch can be
    replayed by passing it back in.
    """
    check_batch_size(count)
    resolved = resolve_seed(seed)
    orders = build_batch(mode, count, resolved)

    counts: Counter[str] = Counter()
    failures: list[str] = []
    accepted_cents = 0

    try:
        with make_client() as client:
            for order in orders:
                response = client.post("/orders", json=order)
                counts[str(response.status_code)] += 1
                if response.status_code == 201:
                    accepted_cents += response.json()["total_cents"]
                elif len(failures) < 5:
                    failures.append(f"{response.status_code} {response.json()['detail']}")
    except httpx.HTTPError as exc:
        raise ToolError(f"cannot reach the order API at {SETTINGS.api_url}: {exc}") from exc

    accepted = counts.get("201", 0)
    return SendResult(
        mode=mode,
        seed=resolved,
        requested=count,
        accepted=accepted,
        rejected=count - accepted,
        status_counts=dict(counts),
        accepted_total_cents=accepted_cents,
        sample_failures=failures,
    )


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

Detailed breakdown

  • The three tools split by what they touch. preview_order never leaves the process, check_target reads the API, and send_orders writes to it. The annotations say so, which is what lets a client auto-approve the first two and confirm the third.
  • destructiveHint=False on send_orders is deliberate and worth reading carefully: posting orders creates records but removes and overwrites nothing. Reserve destructiveHint for irreversible loss, or clients learn to ignore it.
  • openWorldHint separates the tool that stays local from the two that reach an external system, which is the honest signal for a client deciding how much to trust a result.
  • make_client exists as a named function purely as a seam. Tests replace it to drive the API in-process, so the suite needs no port and no server process.
  • resolve_seed draws from random.SystemRandom when the caller omits a seed, and the result reports it back. An agent that stumbles onto a failing batch can hand you the seed and you can replay the identical payloads, which is the difference between a reproducible bug and an anecdote.
  • check_batch_size raises ToolError rather than clamping. A ToolError becomes an error result the model can read and act on, so it retries with a smaller batch instead of silently receiving less traffic than it asked for.
  • status_counts is keyed by string because JSON object keys are strings. Typing it dict[int, int] would produce a schema that disagrees with the payload the client actually receives.
  • sample_failures is capped at five. A rejected batch of 100 would otherwise return 100 error strings into a model’s context for no additional information.
  • The httpx.HTTPError handler converts a connection failure into a readable ToolError. Without it the model sees a masked internal error and has no way to tell that the API is simply not running.
  • mask_error_details=True hides the text of unexpected exceptions. Messages raised deliberately as ToolError are always delivered.
  • The synthetic://catalog resource exposes the sellable rows, so an agent can look up what exists rather than guessing SKUs. It reuses sellable, so the out-of-stock row is filtered in one place.

Step 7: Drive it in-process

The fastest way to exercise the tools is FastMCP’s in-memory client, which talks to the server object directly with no subprocess involved.

Create the file

touch client.py

Add the code: client.py

"""Drive the server in-process: list the tools, preview one order, send a batch.

Needs the order API running (`make api` in another terminal).
"""

import asyncio
import json

from fastmcp import Client

from server import mcp


async def main() -> None:
    async with Client(mcp) as client:
        print("tools:")
        for tool in await client.list_tools():
            hints = tool.annotations
            kind = "read-only" if hints.readOnlyHint else "writes"
            print(f"  {tool.name:<14} {kind:<10} {hints.title}")

        status = (await client.call_tool("check_target", {})).data
        print(f"\ntarget: {status.api_url} reachable={status.reachable} skus={status.skus}")
        if not status.reachable:
            print("start the order API with `make api`, then rerun this demo")
            return

        preview = (await client.call_tool("preview_order", {"seed": 1337})).data
        print(f"\npreview (seed {preview.seed}): {json.dumps(preview.order)}")

        result = (await client.call_tool("send_orders", {"count": 25, "seed": 1337})).data
        print(
            f"\nsent {result.requested} {result.mode} orders (seed {result.seed}): "
            f"{result.status_counts} accepted_total_cents={result.accepted_total_cents}"
        )

        bad = (await client.call_tool("send_orders", {"count": 3, "mode": "simple", "seed": 1337})).data
        print(f"\nsent {bad.requested} simple orders: {bad.status_counts}")
        for failure in bad.sample_failures:
            print(f"  {failure}")


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

Detailed breakdown

  • Client(mcp) takes the server object rather than a command, so the whole exchange happens in one process. That is the right tool for iterating on tool design and the wrong tool for proving deployment works, which Step 8 covers.
  • .data gives the deserialized return value, so status.reachable and result.status_counts are ordinary Python attributes. It is not the server’s SendResult instance, even here: the client rebuilds it from the output schema. Step 8 shows what that costs you.
  • The demo checks reachability before sending and exits with an instruction instead of a traceback when the API is down.
  • Running the naive mode last shows both halves of the story: accepted traffic, then the rejection path with the API’s own error strings passed through.

Run it

Start the order API in one terminal:

uv run uvicorn api.main:app --reload

Then run the demo in another:

uv run python client.py

tools:
  check_target   read-only  Check the order API
  preview_order  read-only  Preview a synthetic order
  send_orders    writes     Send synthetic orders

target: http://127.0.0.1:8000 reachable=True skus=8

preview (seed 1337): {"request_id": "643cb56d-4ec1-4fc6-bee2-9f53ebf644bb", "customer_id": "CUST-0005", "channel": "partner", "currency": "USD", "lines": [{"sku": "SKU-6300", "quantity": 3, "unit_price_cents": 2450}, {"sku": "SKU-5510", "quantity": 4, "unit_price_cents": 1850}, {"sku": "SKU-6301", "quantity": 3, "unit_price_cents": 3900}], "total_cents": 26450}

sent 25 seeded orders (seed 1337): {'201': 25} accepted_total_cents=715350

sent 3 simple orders: {'422': 3}
  422 ['unknown customer C-5093', 'unknown sku SKU-9757', 'unknown sku SKU-6393', 'unknown sku SKU-1830']
  422 ['unknown customer C-8975', 'unknown sku SKU-8549']
  422 ['unknown customer C-5035', 'unknown sku SKU-5650', 'unknown sku SKU-6612', 'unknown sku SKU-0935']

The preview at seed 1337 is the same order the previous article generated at that seed, which is the point of threading the seed through: the MCP layer changed how a batch is triggered, not what it contains. As before, those exact values are what seed 1337 produces on CPython 3.12; a different Python could produce different ones, and nothing here depends on the specific values.

Step 8: Prove the stdio path

An in-memory client skips process launch, transport framing, and JSON round-tripping. Those are exactly the parts that break when a real client tries to start your server.

Create the file

touch scripts/smoke_stdio.py

Add the code: scripts/smoke_stdio.py

"""Exercise the server the way a real MCP client does: as a stdio subprocess.

The in-memory client in client.py imports the server object directly, which
skips process launch, transport framing, and JSON round-tripping. This script
spawns `python server.py` and talks to it over stdio, so a pass here means the
command an MCP client is configured with actually works.
"""

import asyncio
import json
import sys
from pathlib import Path

from fastmcp import Client

SERVER = Path(__file__).resolve().parent.parent / "server.py"


async def main() -> int:
    async with Client(str(SERVER)) as client:
        tools = sorted(t.name for t in await client.list_tools())
        print(f"tools: {tools}")

        status = (await client.call_tool("check_target", {})).data
        print(f"target: {status.api_url} reachable={status.reachable}")
        if not status.reachable:
            print("order API is not running; start it with `make api`", file=sys.stderr)
            return 1

        result = await client.call_tool("send_orders", {"count": 5, "seed": 1337})
        # Any client rebuilds structured output from the tool's output schema, so
        # `data` is a synthetic model named Root, not the server's SendResult —
        # on the in-memory transport as well as this one.
        print(f"data type: {type(result.data).__name__}")
        print(f"structured: {json.dumps(result.structured_content)}")
        assert result.data.accepted == 5, result.structured_content
        print("OK")
        return 0


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

Detailed breakdown

  • Passing a .py path to Client makes FastMCP infer stdio transport and launch the file as a subprocess, which is the same mechanism an MCP client uses.
  • SERVER is resolved from __file__, so the script works regardless of the directory it is invoked from.
  • The return code is meaningful: 1 when the API is unreachable, 0 on success. That makes the script usable as a CI gate rather than something a human has to read.
  • The comment on result.data records a real cross-transport difference, covered in the run below. It is the kind of detail that silently breaks a client script written against the in-memory API.

Run it

With the API still running:

uv run python scripts/smoke_stdio.py

tools: ['check_target', 'preview_order', 'send_orders']
target: http://127.0.0.1:8000 reachable=True
data type: Root
structured: {"mode": "seeded", "seed": 1337, "requested": 5, "accepted": 5, "rejected": 0, "status_counts": {"201": 5}, "accepted_total_cents": 73550, "sample_failures": []}
OK

data type: Root is worth pausing on, and it is not a property of stdio. Every client rebuilds structured output from the tool’s advertised output schema, so result.data is a synthetic class FastMCP generates, named Root, on the in-memory transport in Step 7 just as much as here. Attribute access works (result.data.accepted is 5), but methods of the server’s class do not exist: result.data.model_dump() raises AttributeError either way. Read result.structured_content when you want the raw dict.

The server’s own SendResult only exists on the server side of the call, which is why make send can call model_dump() on it: that target invokes the tool function directly and never crosses a client at all.

Step 9: Register the server with an MCP client

The command an MCP client runs needs an absolute path, which differs per machine. Committing one would break for everyone else, so the config is generated from a tracked template.

Create the file

touch .mcp.json.example

Add the code: .mcp.json.example

{
  "mcpServers": {
    "synthetic-orders": {
      "command": "uv",
      "args": ["run", "--directory", "__PROJECT_DIR__", "python", "server.py"],
      "env": {
        "ORDER_API_URL": "http://127.0.0.1:8000",
        "MAX_BATCH": "100"
      }
    }
  }
}

Detailed breakdown

  • __PROJECT_DIR__ is substituted by make config (Step 11), which writes the gitignored .mcp.json. The template is tracked; the rendered file is not.
  • uv run --directory <path> is what makes the absolute path necessary. An MCP client launches the server as a child process and it inherits the client’s working directory, not the project’s, so a relative path resolves somewhere unpredictable.
  • env is where the guardrails are set. This is the file that decides which API the traffic hits and how large a batch may be, which is precisely why those are not tool parameters.
  • Pointing ORDER_API_URL at a staging host is the intended way to retarget the server. It is a config change by whoever runs it, not a decision a prompt can make.

Register it

make config

sed 's|__PROJECT_DIR__|/path/to/synthetic-orders-mcp|' .mcp.json.example > .mcp.json
wrote .mcp.json for /path/to/synthetic-orders-mcp

A project-scoped .mcp.json is picked up by Claude Code when it starts in that directory, and it waits for approval before running:

claude mcp list

synthetic-orders: uv run --directory /path/to/synthetic-orders-mcp python server.py - ⏸ Pending approval (run `claude` to approve)

Approve it by running claude in the project directory. The alternative is to register it directly, which connects immediately:

claude mcp add synthetic-orders --scope local \
  --env ORDER_API_URL=http://127.0.0.1:8000 --env MAX_BATCH=100 \
  -- uv run --directory "$(pwd)" python server.py

Added stdio MCP server synthetic-orders with command: uv run --directory /path/to/synthetic-orders-mcp python server.py to local config

Pick one of the two. Doing both registers the same name in two scopes, and claude mcp list will say so:

├ Server "synthetic-orders" is defined in multiple scopes with different endpoints: project (...), local (...).
└ Keep the correct endpoint and remove the others: `claude mcp remove synthetic-orders -s project` or `claude mcp remove synthetic-orders -s local`

With the API running and the server approved, an agent can now be asked for traffic in plain language: “check the order API, then send 25 synthetic orders and tell me the status breakdown.” It will call check_target, then send_orders, and read back the counts.

Step 10: Test the control surface

The tests assert the design contract, not just that the tools run.

Create the file

pyproject.toml already exists from uv init and the two uv add commands. Append the [tool.pytest.ini_options] block, and replace the placeholder description that uv init wrote.

Add the code: pyproject.toml

[project]
name = "synthetic-orders-mcp"
version = "0.1.0"
description = "Trigger synthetic order traffic from an MCP server"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "fastapi>=0.141.1",
    "fastmcp>=3.4.6",
    "httpx>=0.28.1",
    "uvicorn[standard]>=0.52.1",
]

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

[tool.pytest.ini_options]
pythonpath = ["."]
testpaths = ["tests"]
asyncio_mode = "auto"

Detailed breakdown

  • pythonpath = ["."] lets the tests import server, api.main, and synth without installing the project. Without it, collection fails with ModuleNotFoundError: No module named 'server'.
  • asyncio_mode = "auto" runs every async def test_ without a per-test decorator, which matters because the MCP client API is async throughout.
  • testpaths = ["tests"] keeps collection out of .venv/ and scripts/.

Create the file

touch tests/test_server.py

Add the code: tests/test_server.py

"""Tests for the synthetic-order MCP server.

These assert the control surface, not just that the tools run: the annotations
each tool advertises, the guardrails on batch size and target selection, seed
round-tripping, and what happens when the order API is unreachable.
"""

import httpx
import pytest
from fastapi.testclient import TestClient
from fastmcp import Client
from fastmcp.exceptions import ToolError

import server
from api.main import app


@pytest.fixture(autouse=True)
def in_process_api(monkeypatch):
    """Route the server's HTTP client at the API in-process, no port needed.

    `TestClient` subclasses `httpx.Client` and drives the ASGI app directly, so
    it drops into the same seam `make_client` exists for. A plain
    `httpx.ASGITransport` cannot: it is async-only, and `send_orders` uses a
    synchronous client.
    """
    monkeypatch.setattr(server, "make_client", lambda: TestClient(app))


async def test_annotations_separate_reads_from_writes():
    async with Client(server.mcp) as client:
        tools = {t.name: t.annotations for t in await client.list_tools()}
        assert tools["preview_order"].readOnlyHint is True
        assert tools["check_target"].readOnlyHint is True
        assert tools["send_orders"].readOnlyHint is False
        assert tools["send_orders"].idempotentHint is False
        # Sending orders creates records; it does not remove or overwrite any.
        assert tools["send_orders"].destructiveHint is False
        # preview_order never leaves the process; the other two reach the API.
        assert tools["preview_order"].openWorldHint is False
        assert tools["send_orders"].openWorldHint is True


async def test_no_tool_accepts_a_target_url():
    """The traffic target is server configuration, never a model-supplied value."""
    async with Client(server.mcp) as client:
        for tool in await client.list_tools():
            properties = (tool.inputSchema or {}).get("properties", {})
            assert not [p for p in properties if "url" in p.lower() or "host" in p.lower()]


async def test_check_target_reports_the_seed_counts():
    async with Client(server.mcp) as client:
        status = (await client.call_tool("check_target", {})).data
        assert status.reachable is True
        assert (status.skus, status.customers) == (8, 5)


async def test_preview_does_not_send():
    async with Client(server.mcp) as client:
        before = (await client.call_tool("check_target", {})).data
        preview = (await client.call_tool("preview_order", {"seed": 1337})).data
        assert preview.order["customer_id"] == "CUST-0005"
        assert preview.total_cents == 26450
        assert preview.line_count == 3
        after = (await client.call_tool("check_target", {})).data
        assert before.skus == after.skus


async def test_seeded_batch_is_accepted_and_replayable():
    async with Client(server.mcp) as client:
        first = (await client.call_tool("send_orders", {"count": 25, "seed": 1337})).data
        assert first.status_counts == {"201": 25}
        assert first.accepted == 25 and first.rejected == 0
        second = (await client.call_tool("send_orders", {"count": 25, "seed": first.seed})).data
        assert second.accepted_total_cents == first.accepted_total_cents


async def test_omitted_seed_is_reported_back():
    async with Client(server.mcp) as client:
        result = (await client.call_tool("send_orders", {"count": 3})).data
        replay = (await client.call_tool("send_orders", {"count": 3, "seed": result.seed})).data
        assert replay.accepted_total_cents == result.accepted_total_cents


async def test_simple_mode_drills_the_rejection_path():
    async with Client(server.mcp) as client:
        result = (await client.call_tool("send_orders", {"count": 5, "mode": "simple", "seed": 1337})).data
        assert result.status_counts == {"422": 5}
        assert result.accepted == 0 and result.accepted_total_cents == 0
        assert any("unknown customer" in f for f in result.sample_failures)


async def test_batch_size_is_capped():
    async with Client(server.mcp) as client:
        with pytest.raises(ToolError, match="exceeds the server's MAX_BATCH"):
            await client.call_tool("send_orders", {"count": server.SETTINGS.max_batch + 1})
        with pytest.raises(ToolError, match="at least 1"):
            await client.call_tool("send_orders", {"count": 0})


async def test_unreachable_api_is_a_readable_error(monkeypatch):
    def broken_client() -> httpx.Client:
        def fail(request: httpx.Request) -> httpx.Response:
            raise httpx.ConnectError("connection refused", request=request)

        return httpx.Client(transport=httpx.MockTransport(fail), base_url="http://127.0.0.1:9")

    monkeypatch.setattr(server, "make_client", broken_client)
    async with Client(server.mcp) as client:
        status = (await client.call_tool("check_target", {})).data
        assert status.reachable is False
        with pytest.raises(ToolError, match="cannot reach the order API"):
            await client.call_tool("send_orders", {"count": 1})


async def test_catalog_resource_exposes_only_sellable_rows():
    async with Client(server.mcp) as client:
        contents = await client.read_resource("synthetic://catalog")
        rows = contents[0].text
        assert "SKU-5510" in rows
        assert "SKU-7101" not in rows

Detailed breakdown

  • The in_process_api fixture is where the make_client seam pays off. Starlette’s TestClient subclasses httpx.Client and drives the ASGI app directly, so the whole suite runs with no port and no server process. The docstring records why the obvious alternative fails: httpx.ASGITransport is async-only, and send_orders uses a synchronous client, so a sync httpx.Client wrapped around it raises AttributeError: 'ASGITransport' object has no attribute '__enter__'.
  • test_no_tool_accepts_a_target_url reads the advertised input schemas and fails if any tool grows a URL or host parameter. The guardrail from Step 5 is a design decision, and this is what stops a later refactor from quietly undoing it.
  • test_annotations_separate_reads_from_writes pins the hints a client uses to decide what to auto-approve. Flipping readOnlyHint on send_orders would be a security-relevant change, so it should break a test.
  • test_preview_does_not_send proves the read-only claim rather than trusting the annotation, by checking the API’s state around the call.
  • test_seeded_batch_is_accepted_and_replayable and test_omitted_seed_is_reported_back cover both halves of the seed contract: an explicit seed reproduces a batch, and an omitted one is reported so it can be reused.
  • test_batch_size_is_capped reads server.SETTINGS.max_batch instead of hard-coding 100, so the test follows the configured limit.
  • test_unreachable_api_is_a_readable_error uses httpx.MockTransport to raise a connection error deterministically, with no dependence on a port being closed. It asserts the two different behaviors: check_target reports reachable=False, while send_orders raises.
  • The resource test asserts on the raw text, which is the simplest way to check that the out-of-stock row never reaches a client.

Run the suite

uv run pytest -q

..........                                                               [100%]
=============================== warnings summary ===============================
.venv/lib/python3.12/site-packages/fastapi/testclient.py:1
  .../fastapi/testclient.py:1: StarletteDeprecationWarning: Using `httpx` with `starlette.testclient` is deprecated; install `httpx2` instead.
    from starlette.testclient import TestClient as TestClient  # noqa

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
10 passed, 1 warning in 0.41s

The suite needs no running API. The reported duration varies by machine.

Step 11: Wrap it in a Makefile

Create the file

touch Makefile

Add the code: Makefile

.DEFAULT_GOAL := help
ORDER_API_URL ?= http://127.0.0.1:8000
COUNT ?= 25
SEED ?= 1337

.PHONY: help install api serve demo smoke send config test clean

help: ## Show this help screen
	@echo "Synthetic order MCP server"
	@echo ""
	@echo "Targets:"
	@grep -E '^[a-zA-Z_-]+:.*?## ' $(MAKEFILE_LIST) \
		| awk 'BEGIN {FS = ":.*?## "}; {printf "  %-10s %s\n", $$1, $$2}'
	@echo ""
	@echo "Variables: ORDER_API_URL=$(ORDER_API_URL) COUNT=$(COUNT) SEED=$(SEED)"

install: ## Sync dependencies with uv
	uv sync

api: ## Start the order API this server sends traffic to
	uv run uvicorn api.main:app --reload

serve: ## Run the MCP server over stdio
	uv run python server.py

demo: ## Drive the server in-process (needs `make api` running)
	uv run python client.py

smoke: ## Launch the server as a stdio subprocess and send a batch
	uv run python scripts/smoke_stdio.py

send: ## Fire one batch through the MCP tool without a client app
	uv run python -c "import json, server; print(json.dumps(server.send_orders($(COUNT), 'seeded', $(SEED)).model_dump(), indent=2))"

config: ## Render .mcp.json from the example with this project's absolute path
	sed 's|__PROJECT_DIR__|$(CURDIR)|' .mcp.json.example > .mcp.json
	@echo "wrote .mcp.json for $(CURDIR)"

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

clean: ## Remove caches and bytecode
	rm -rf .pytest_cache **/__pycache__

Detailed breakdown

  • .DEFAULT_GOAL := help makes a bare make print the target list, and the help target greps its own ## comments so adding a target adds a help entry.
  • api and serve are separate processes on purpose: one is the system under test, the other is the thing that drives it.
  • demo and smoke are the two client paths, in-process and stdio. Run smoke before wiring the server into a real client, since it exercises the same launch command.
  • send calls the tool function directly, skipping MCP entirely. It is the fastest way to confirm the generator and the API agree when a tool call is misbehaving, because it removes the protocol from the picture.
  • config renders .mcp.json with $(CURDIR), which is how the absolute path gets in without anyone hand-editing JSON.
  • ORDER_API_URL is echoed in the help screen so the target is visible without reading the Makefile. Its default is a Make variable, which is not exported, so make send on its own uses whatever the server’s environment already holds. A command-line override behaves differently: GNU make exports variables set on the command line, so make send ORDER_API_URL=http://127.0.0.1:9 really does retarget the batch and fails with a connection error.

Verify the default target

make

Synthetic order MCP server

Targets:
  help       Show this help screen
  install    Sync dependencies with uv
  api        Start the order API this server sends traffic to
  serve      Run the MCP server over stdio
  demo       Drive the server in-process (needs `make api` running)
  smoke      Launch the server as a stdio subprocess and send a batch
  send       Fire one batch through the MCP tool without a client app
  config     Render .mcp.json from the example with this project's absolute path
  test       Run the test suite
  clean      Remove caches and bytecode

Variables: ORDER_API_URL=http://127.0.0.1:8000 COUNT=25 SEED=1337

Step 12: Full validation run

make install
make test

In one terminal:

make api

In another:

make smoke
make send COUNT=5 SEED=42

{
  "mode": "seeded",
  "seed": 42,
  "requested": 5,
  "accepted": 5,
  "rejected": 0,
  "status_counts": {
    "201": 5
  },
  "accepted_total_cents": 109450,
  "sample_failures": []
}

Then confirm the guardrails hold. The batch cap:

MAX_BATCH=5 uv run python -c "
import server
from fastmcp.exceptions import ToolError
try:
    server.send_orders(count=6)
except ToolError as e:
    print('ToolError:', e)
"

ToolError: count 6 exceeds the server's MAX_BATCH of 5; send smaller batches or raise MAX_BATCH in the server environment

And the target, which follows the environment rather than any argument:

ORDER_API_URL=http://127.0.0.1:9 uv run python -c "
import server
print(server.check_target().model_dump())
"

{'api_url': 'http://127.0.0.1:9', 'reachable': False, 'skus': None, 'customers': None}

Port 9 is the discard port, so nothing is listening there. The server reports the target it was configured with and that it could not reach it, which is the behavior an agent needs in order to say something useful instead of retrying blindly.

Troubleshooting

ModuleNotFoundError: No module named 'server' when running pytest. The [tool.pytest.ini_options] block is missing or lost pythonpath = ["."]. Confirm pytest’s header says configfile: pyproject.toml.

Tests are collected but skipped, or fail with “async def functions are not natively supported”. asyncio_mode = "auto" is missing, or pytest-asyncio was not installed as a dev dependency.

AttributeError: 'ASGITransport' object has no attribute '__enter__'. httpx.ASGITransport is async-only. Use fastapi.testclient.TestClient, which subclasses httpx.Client and works with the synchronous client send_orders uses.

AttributeError: 'Root' object has no attribute 'model_dump'. Expected from any client, in-memory or stdio: result.data is rebuilt from the output schema, so it is not your SendResult class. Read result.structured_content for the dict, or access the fields directly. Calling the tool function directly, as make send does, returns the real model.

The server shows “Pending approval” in claude mcp list. That is the normal state for a project-scoped .mcp.json. Run claude in the project directory and approve it.

claude mcp list warns the server is defined in multiple scopes. Both .mcp.json and claude mcp add registered the same name. Remove one with claude mcp remove synthetic-orders -s project or -s local.

Every batch comes back 422 in seeded mode. The server and the API are reading different seed data. Both resolve data/ relative to their own __file__, so this means the files were edited in one copy of the project only.

cannot reach the order API at http://127.0.0.1:8000. The API is not running, or ORDER_API_URL points somewhere else. Check with make api and curl -s http://127.0.0.1:8000/health.

Recap

  • The generators are unchanged from the previous article. What is new is a control surface: check_target, preview_order, and send_orders, with annotations that let a client auto-approve the reads and confirm the write.
  • The traffic target comes from ORDER_API_URL in the server’s environment and is not a tool parameter, so a prompt cannot aim the generator at another host. A test reads the advertised schemas and fails if any tool grows a URL argument.
  • MAX_BATCH caps the blast radius, and exceeding it raises a ToolError the model can read and act on rather than silently sending less traffic.
  • Every batch reports the seed it used, so an agent that finds a failing batch can hand back something you can replay exactly.
  • Any MCP client rebuilds a structured result from the tool’s output schema, so result.data is a synthetic Root, not the server’s SendResult, on the in-memory transport as well as stdio. Attribute access is portable; model_dump() is not.

Next improvements

  • Add a stop control and run batches in the background, so an agent can start sustained traffic and end it, rather than blocking on one synchronous loop.
  • Report latency percentiles alongside status counts, which turns the tool from a correctness drill into a smoke-level load check.
  • Add an invalid_order mode that mutates one field of a valid payload, giving each API rule its own negative-path trigger.
  • Emit progress notifications for long batches so a client can show them as they run; see Stream Progress from a Long-Running FastMCP Tool on macOS.
  • Put the server behind HTTP with auth if it should be shared, which turns ORDER_API_URL into per-deployment configuration and makes the batch cap a per-caller quota.