Testing MCP servers doesn't need a running server, a model or a network. Promptise Foundry's TestClient pushes each call through the same pipeline the live server uses: input validation, authentication, guards, middleware, your handler and the structured error envelope. Plain pytest tests can then check exactly what an agent would get back. This guide tests a small order-support MCP server with 35 tests covering tools, tokens and roles, middleware, resources and prompts, error envelopes and a background job queue, two of them over real HTTP, and runs them with a GitHub Actions workflow. The whole suite runs in about a second and makes no model calls. Every test ran against Promptise Foundry 1.3.3 installed from PyPI, and every output below is pasted from those runs.
How do you test an MCP server without running it?
Build the server, wrap it in a TestClient that carries the caller's credentials, and call tools the way an MCP client would. The client returns the content the server would send, so a refused call comes back as the structured error the model reads, not as an exception:
async def test_an_agent_cannot_refund(agent, payments):
result = await agent.call_tool("refund_order", {"order_id": "A-1001", "amount": 2500})
error = error_of(result)
assert error["code"] == "ACCESS_DENIED"
assert error["message"] == "Requires any of roles [supervisor], but client has [agent]"
assert payments.refunds == []agent is a fixture: a TestClient holding a support bot's signed token. payments is a fake payment client, so the last line proves no money moved. Seven tool tests like this one run in about a tenth of a second:
pytest -q tests/test_tools.py....... [100%]
7 passed in 0.11s[02]
What TestClient runs, and what it skips
TestClient replays the server's request pipeline in your test process, without a transport. Most of what can go wrong in an MCP server lives in that pipeline, so most tests belong here. A few behaviours only exist on the wire, and the table says which:
Behaviour | TestClient | Live HTTP server |
|---|---|---|
Input validation against the tool's schema | Yes | Yes |
Dependency injection and RequestContext | Yes | Yes |
Middleware, in the order you added it | Yes | Yes |
Authentication and guards | Yes | Yes |
Exception handlers and error envelopes | Yes | Yes |
Background tasks | Run before call_tool returns | Run after the response is sent |
The HTTP auth gate (401 before any MCP message) | No | Yes, with require_auth=True |
The isError flag on a failed call | No, parse the JSON | Yes |
Startup and shutdown hooks, such as queue workers | No, start them yourself | Yes |
Host and Origin checks, CORS, MCP sessions | No | Yes |
The plan that follows from it: test the pipeline with TestClient, and add a couple of wire-level tests for the rows marked No. Both are below.
[03]
What you need
Python 3.10 or newer.
Promptise Foundry 1.3.3 or newer, pytest and pytest-asyncio.
No API key. Nothing in this guide calls a model.
pip install "promptise>=1.3.3" pytest pytest-asyncioThe project has one server module and a test folder:
orders-mcp/
├── orders_server.py
├── pyproject.toml
├── .github/workflows/tests.yml
└── tests/
├── conftest.py
├── helpers.py
├── test_tools.py
├── test_auth.py
├── test_errors.py
├── test_middleware.py
├── test_resources_prompts.py
├── test_queue.py
└── test_wire.py[04]
Set up pytest for an MCP server, step by step
Build the server in a function
The server under test is an order-support MCP server for a shop's agents. Agents look orders up, supervisors refund them, a policy is published as a resource, a prompt drafts the customer reply, and a background job reconciles payments. Notice the shape: everything is built inside create_app(), not at import time. If these server features are new to you, Build a Production MCP Server in Python (Step by Step) introduces them one at a time; this guide is about testing them.
import asyncio
import copy
import os
import secrets
from dataclasses import dataclass
from typing import Annotated
from promptise.mcp.server import (
AuthMiddleware,
JWTAuth,
MCPQueue,
MCPServer,
RateLimitMiddleware,
RequestContext,
ToolError,
)
from pydantic import Field
# A stand-in for your order database. Amounts are in cents.
SEED_ORDERS = {
"A-1001": {"customer": "acme", "total": 12_900, "status": "paid"},
"A-1002": {"customer": "acme", "total": 4_500, "status": "refunded"},
"G-2001": {"customer": "globex", "total": 89_000, "status": "paid"},
}
OrderId = Annotated[
str, Field(description='An order ID, for example "A-1001".', pattern=r"^[A-Z]-\d{4}$")
]
class PaymentGatewayDown(Exception):
"""The payment provider can't be reached."""
class Payments:
"""Your payment provider's client. Tests pass a fake instead."""
async def refund(self, order_id: str, amount: int) -> str:
return f"re_{secrets.token_hex(4)}" # call the provider's refund API here
class ReadOnlyMode:
"""Middleware: refuse destructive tools while writes are frozen."""
def __init__(self) -> None:
self.enabled = False
async def __call__(self, ctx: RequestContext, call_next):
hints = getattr(ctx.state.get("tool_def"), "annotations", None)
if self.enabled and hints is not None and hints.destructive_hint:
raise ToolError(
"Writes are paused while an incident is open.",
code="READ_ONLY",
retryable=True,
suggestion="Try again later, or ask a person to make the change.",
)
return await call_next(ctx)
@dataclass
class OrdersApp:
server: MCPServer
queue: MCPQueue
read_only: ReadOnlyMode
def create_app(*, jwt_secret: str, payments: Payments | None = None) -> OrdersApp:
"""Build a fresh server with its own data, rate limits and queue."""
orders = copy.deepcopy(SEED_ORDERS)
payments = payments or Payments()
read_only = ReadOnlyMode()
server = MCPServer("orders", require_auth=True)
# Middleware runs in the order it's added. Authenticate first,
# so the rate limiter can keep one bucket per caller.
server.add_middleware(AuthMiddleware(JWTAuth(secret=jwt_secret, audience="orders-mcp")))
server.add_middleware(RateLimitMiddleware(rate_per_minute=60, burst=5))
server.add_middleware(read_only)
@server.exception_handler(PaymentGatewayDown)
async def payment_down(ctx, exc):
message = "The payment provider is unavailable."
return ToolError(message, code="PAYMENT_UNAVAILABLE", retryable=True)
def find(order_id: str, ctx: RequestContext) -> dict:
"""An order the caller's tenant may see, or a ToolError."""
order = orders.get(order_id)
if order is None or order["customer"] != ctx.client.tenant_id:
raise ToolError(f"There is no order {order_id}.", suggestion="Check the order ID.")
return order
@server.tool(read_only_hint=True)
async def get_order(order_id: OrderId, ctx: RequestContext) -> dict:
"""Get one order with its total and status."""
return {"order_id": order_id, **find(order_id, ctx)}
@server.tool(roles=["supervisor"], destructive_hint=True)
async def refund_order(
order_id: OrderId,
amount: Annotated[int, Field(description="Amount to refund, in cents.", gt=0)],
ctx: RequestContext,
) -> dict:
"""Refund part or all of a paid order. Only supervisors can refund."""
order = find(order_id, ctx)
if order["status"] != "paid":
raise ToolError(f"Order {order_id} is {order['status']}, so it can't be refunded.")
if amount > order["total"]:
raise ToolError(
f"The refund is larger than the order total of {order['total']} cents.",
code="REFUND_TOO_LARGE",
details={"order_total": order["total"]},
)
refund_id = await payments.refund(order_id, amount)
order["status"] = "refunded"
return {"order_id": order_id, "refund_id": refund_id, "status": "refunded"}
@server.resource("policy://refunds")
async def refund_policy() -> str:
"""The refund policy agents must follow."""
return "Refunds over 500.00 USD need a second supervisor. Never refund an order twice."
@server.resource_template("orders://{order_id}/summary")
async def order_summary(order_id: str, ctx: RequestContext) -> dict:
"""A one-line summary of an order."""
order = find(order_id, ctx)
return {
"order_id": order_id,
"summary": f"{order['status']}, {order['total'] / 100:.2f} USD",
}
@server.prompt()
async def refund_reply(order_id: str, amount_cents: int) -> str:
"""Draft a customer reply confirming a refund."""
amount = f"{amount_cents / 100:.2f} USD"
return f"Write a short, friendly reply confirming a refund of {amount} for order {order_id}."
queue = MCPQueue(server, max_workers=2)
@queue.job(name="reconcile", timeout=30)
async def reconcile(day: str) -> dict:
"""Reconcile one day's payments against the provider's settlement report."""
await asyncio.sleep(0.05) # stands in for a slow batch job
return {"day": day, "matched": 412, "mismatched": 0}
return OrdersApp(server=server, queue=queue, read_only=read_only)
if __name__ == "__main__":
app = create_app(jwt_secret=os.environ["ORDERS_JWT_SECRET"])
app.server.run(transport="http", host="127.0.0.1", port=8080)Three choices in this file make it testable:
A factory, not a module-level server. Each call to create_app() builds a new server with its own copy of the orders, its own rate-limit buckets and its own queue. Tests that refund an order or burn through a rate limit can't affect the next test.
Collaborators are passed in. Depends() has no FastAPI-style dependency_overrides, so the way to swap the real payment client for a fake is to hand the fake to the factory.
The objects tests need are returned. OrdersApp gives tests the queue, to start its workers, and the read-only switch, to flip it.
Configure pytest for async tests
Every TestClient method is a coroutine, so the tests are async def. pytest-asyncio runs them; in auto mode it does so without a marker on each test:
[project]
name = "orders-mcp"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = ["promptise>=1.3.3"]
[project.optional-dependencies]
test = ["pytest>=8", "pytest-asyncio>=0.24"]
[tool.pytest.ini_options]
asyncio_mode = "auto"
pythonpath = ["."]
testpaths = ["tests"]pythonpath = ["."] lets the tests import orders_server from the project root without installing it. In pytest-asyncio's default strict mode, you'd decorate each test with @pytest.mark.asyncio and each async fixture with @pytest_asyncio.fixture instead.
Write fixtures for the callers
Most MCP server tests are about who is calling. A small fixture factory turns "call this as Maya, a supervisor at Acme" into one line. Tokens are minted with JWTAuth.create_token, a helper Promptise provides for tests, signed with a test-only secret:
import json
from orders_server import Payments
SECRET = "test-only-secret" # signs test tokens; never your real secret
class FakePayments(Payments):
"""Records refunds instead of moving money. Set .fail to make it raise."""
def __init__(self) -> None:
self.refunds: list[tuple[str, int]] = []
self.fail: Exception | None = None
async def refund(self, order_id: str, amount: int) -> str:
if self.fail is not None:
raise self.fail
self.refunds.append((order_id, amount))
return f"re_test_{len(self.refunds)}"
def error_of(result) -> dict:
"""The structured error of a failed tool call, or {} when it succeeded."""
try:
body = json.loads(result[0].text)
except ValueError:
return {}
return body.get("error", {}) if isinstance(body, dict) else {}import pytest
from helpers import SECRET, FakePayments
from orders_server import create_app
from promptise.mcp.server import JWTAuth, TestClient
@pytest.fixture
def payments():
return FakePayments()
@pytest.fixture
def app(payments):
"""A fresh server for every test, so no state leaks from one test to the next."""
return create_app(jwt_secret=SECRET, payments=payments)
@pytest.fixture
def token():
"""Mint a signed test token: token("maya", roles=("supervisor",))."""
issuer = JWTAuth(secret=SECRET, audience="orders-mcp")
def mint(sub, *, tenant="acme", roles=(), expires_in=300):
claims = {"sub": sub, "tenant_id": tenant, "roles": list(roles)}
return issuer.create_token(claims, expires_in=expires_in)
return mint
@pytest.fixture
def client_for(app, token):
"""A TestClient that calls the server as one user."""
def make(sub, **claims):
return TestClient(app.server, meta={"authorization": f"Bearer {token(sub, **claims)}"})
return make
@pytest.fixture
def agent(client_for):
return client_for("support-bot", roles=("agent",))
@pytest.fixture
def supervisor(client_for):
return client_for("maya", roles=("agent", "supervisor"))All fixtures are function-scoped, the pytest default. That's deliberate: app builds a new server per test, and every client a test makes talks to that one server. client_for returns a function rather than a client, so a test can bring in as many callers as it needs, from any tenant, with any roles.
[05]
Test MCP tools and the schema the model sees
A tool has two audiences. The model reads its name, description and input schema; the agent's code gets its result. Test both:
import json
from helpers import error_of
async def test_the_tool_inventory(agent):
"""Everything a model can call. The queue adds five tools of its own."""
names = {tool.name for tool in await agent.list_tools()}
queue = {"queue_submit", "queue_status", "queue_result", "queue_cancel", "queue_list"}
assert names == {"get_order", "refund_order"} | queue
async def test_the_model_sees_described_inputs(agent):
tools = {tool.name: tool for tool in await agent.list_tools()}
order_id = tools["get_order"].inputSchema["properties"]["order_id"]
assert order_id["pattern"] == r"^[A-Z]-\d{4}$"
assert "for example" in order_id["description"]
assert tools["refund_order"].annotations.destructiveHint is True
async def test_get_order(agent):
result = await agent.call_tool("get_order", {"order_id": "A-1001"})
order = json.loads(result[0].text)
assert order == {"order_id": "A-1001", "customer": "acme", "total": 12900, "status": "paid"}
async def test_a_tenant_cannot_read_another_tenants_order(client_for):
globex = client_for("gina", tenant="globex", roles=("agent",))
result = await globex.call_tool("get_order", {"order_id": "A-1001"})
assert error_of(result)["message"] == "There is no order A-1001."
async def test_a_supervisor_can_refund(supervisor, payments):
result = await supervisor.call_tool("refund_order", {"order_id": "A-1001", "amount": 2500})
assert json.loads(result[0].text)["status"] == "refunded"
assert payments.refunds == [("A-1001", 2500)]
async def test_an_agent_cannot_refund(agent, payments):
result = await agent.call_tool("refund_order", {"order_id": "A-1001", "amount": 2500})
error = error_of(result)
assert error["code"] == "ACCESS_DENIED"
assert error["message"] == "Requires any of roles [supervisor], but client has [agent]"
assert payments.refunds == []
async def test_an_order_is_refunded_only_once(supervisor):
first = await supervisor.call_tool("refund_order", {"order_id": "A-1001", "amount": 100})
second = await supervisor.call_tool("refund_order", {"order_id": "A-1001", "amount": 100})
assert error_of(first) == {}
assert error_of(second)["message"] == "Order A-1001 is refunded, so it can't be refunded."The first test is an inventory: the exact set of tools a model can call. It's the cheapest test in the suite and catches more than it seems. Adding a job queue to this server added five tools, queue_submit to queue_list, which the queue registers on its own. A model sees all of them, so a test should say so, and should fail when a tool appears or disappears without anyone deciding it should.
The second test pins the schema. The pattern and the description come from the Annotated type in the server, and destructiveHint from destructive_hint=True. A client can use that hint to ask a person before calling the tool. The tenant test checks the rule that matters most in a shared server: another tenant's order looks exactly like one that doesn't exist.
[06]
Test authentication and guards
require_auth=True makes every tool authenticate, and AuthMiddleware puts the verified identity on ctx.client. Test the refusals as carefully as the happy path. A refusal you have never seen is a refusal you're trusting on faith:
import pytest
from helpers import SECRET, error_of
from promptise.mcp.server import JWTAuth, TestClient
ARGS = {"order_id": "A-1001"}
FORGER = JWTAuth(secret="not-our-secret", audience="orders-mcp")
OTHER_SERVER = JWTAuth(secret=SECRET, audience="billing-mcp")
def bearer(token):
return {"authorization": f"Bearer {token}"}
async def test_no_token(app):
result = await TestClient(app.server).call_tool("get_order", ARGS)
assert error_of(result) == {
"code": "AUTHENTICATION_ERROR",
"message": "Missing authentication token",
"retryable": False,
}
@pytest.mark.parametrize(
"kind, message",
[
("expired", "Token expired"),
("forged", "Invalid JWT signature"),
("other-audience", "Token was not issued for this server (audience mismatch)"),
],
ids=["expired", "forged", "other-audience"],
)
async def test_bad_tokens_are_refused(app, token, kind, message):
claims = {"sub": "maya", "tenant_id": "acme"}
bad = {
"expired": lambda: token("maya", expires_in=-60),
"forged": lambda: FORGER.create_token(claims),
"other-audience": lambda: OTHER_SERVER.create_token(claims),
}[kind]()
result = await TestClient(app.server, meta=bearer(bad)).call_tool("get_order", ARGS)
assert error_of(result)["message"] == message
async def test_headers_apply_to_one_call(app, token):
"""Per-call headers override the client's meta, for that call only."""
client = TestClient(app.server)
signed_in = await client.call_tool("get_order", ARGS, headers=bearer(token("maya")))
anonymous = await client.call_tool("get_order", ARGS)
assert error_of(signed_in) == {}
assert error_of(anonymous)["code"] == "AUTHENTICATION_ERROR"Each bad token gets the exact message JWTAuth returns, so a test fails if the server starts refusing for a different reason. The last test shows the headers= argument: headers passed to one call_tool override the client's meta for that call only. That's how the live server behaves too, where identity is read from the HTTP request that carries each call.
[07]
Test the error envelopes your agents read
When a call fails, the model doesn't get a stack trace. It gets a JSON envelope with a code, a message, a retryable flag and, sometimes, a suggestion and details. Those fields decide whether an agent fixes its arguments, retries, or gives up, so they're part of your server's contract:
import pytest
from helpers import error_of
from orders_server import PaymentGatewayDown
REFUND = {"order_id": "A-1001", "amount": 100}
@pytest.mark.parametrize(
"tool, args, code, retryable",
[
("get_order", {"order_id": "1001"}, "VALIDATION_ERROR", True),
("refund_order", {"order_id": "A-1001", "amount": 0}, "VALIDATION_ERROR", True),
("refund_order", {"order_id": "A-1001", "amount": 99_999}, "REFUND_TOO_LARGE", False),
("get_order", {"order_id": "A-9999"}, "TOOL_ERROR", False),
("cancel_order", {"order_id": "A-1001"}, "TOOL_NOT_FOUND", None),
],
ids=["bad-order-id", "zero-amount", "too-large", "unknown-order", "unknown-tool"],
)
async def test_error_envelopes(supervisor, tool, args, code, retryable):
error = error_of(await supervisor.call_tool(tool, args))
assert error["code"] == code
assert error.get("retryable") == retryable
async def test_details_tell_the_model_what_to_fix(supervisor):
too_much = {"order_id": "A-1001", "amount": 99_999}
error = error_of(await supervisor.call_tool("refund_order", too_much))
assert error["details"] == {"order_total": 12900}
async def test_a_mapped_exception_becomes_a_retryable_error(supervisor, payments):
payments.fail = PaymentGatewayDown("connect timeout")
error = error_of(await supervisor.call_tool("refund_order", REFUND))
assert error == {
"code": "PAYMENT_UNAVAILABLE",
"message": "The payment provider is unavailable.",
"retryable": True,
}
async def test_an_unexpected_exception_does_not_leak(supervisor, payments):
payments.fail = RuntimeError("vault at postgres://pay:hunter2@10.0.0.7 is down")
result = await supervisor.call_tool("refund_order", REFUND)
assert error_of(result) == {
"code": "INTERNAL_ERROR",
"message": "An internal error occurred.",
"retryable": False,
}
assert "hunter2" not in result[0].textThe parametrized test is a table of the server's failure modes. Here is what the model actually receives for two of them, a refund larger than the order and a malformed order ID:
{
"error": {
"code": "REFUND_TOO_LARGE",
"message": "The refund is larger than the order total of 12900 cents.",
"retryable": false,
"details": {
"order_total": 12900
}
}
}
{
"error": {
"code": "VALIDATION_ERROR",
"message": "Invalid input: order_id: String should match pattern '^[A-Z]-\\d{4}$'",
"retryable": true,
"suggestion": "Check the parameter types and constraints in the tool description.",
"details": {
"field_errors": {
"order_id": "String should match pattern '^[A-Z]-\\d{4}$'"
}
}
}
}What each kind of failure tells the agent:
`VALIDATION_ERROR` is retryable. The arguments didn't match the schema, and the agent can fix them. Your handler never ran.
A `ToolError` is not retryable by default. Give it a specific code and details when the agent can act on them, like REFUND_TOO_LARGE with the order total.
An exception handler turns your own exception into an envelope. @server.exception_handler(PaymentGatewayDown) maps an outage to PAYMENT_UNAVAILABLE with retryable: true.
Anything else becomes `INTERNAL_ERROR`, with a fixed message. The original text, here a connection string with a password, goes to your server log, never to the model. The last test checks that it didn't leak.
[08]
Test middleware, including the order it runs in
Middleware wraps every call: tools, resource reads and prompt requests. It runs in the order you add it, and the first one added runs outermost. Test what each middleware does, and test the behaviour that depends on its position:
from helpers import error_of
ARGS = {"order_id": "A-1001"}
async def test_each_caller_has_its_own_rate_limit(client_for):
maya, sam = client_for("maya"), client_for("sam")
for _ in range(5): # the burst
assert error_of(await maya.call_tool("get_order", ARGS)) == {}
limited = error_of(await maya.call_tool("get_order", ARGS))
assert limited["code"] == "RATE_LIMIT_EXCEEDED"
assert limited["retryable"] is True
assert 0 < limited["details"]["retry_after_seconds"] <= 1
assert error_of(await sam.call_tool("get_order", ARGS)) == {}
async def test_read_only_mode_blocks_writes_but_not_reads(app, supervisor, payments):
app.read_only.enabled = True
refund = await supervisor.call_tool("refund_order", {"order_id": "A-1001", "amount": 100})
assert error_of(refund)["code"] == "READ_ONLY"
assert payments.refunds == []
assert error_of(await supervisor.call_tool("get_order", ARGS)) == {}RateLimitMiddleware keeps one token bucket per caller, keyed on ctx.client_id, so it has to run after AuthMiddleware has set it. The rate-limit test checks that: Maya uses up her burst of five, gets RATE_LIMIT_EXCEEDED with a retry_after_seconds hint, and Sam is unaffected.
To see the test earn its place, swap the two add_middleware lines in create_app() so the rate limiter runs first, and run it again:
> assert error_of(await sam.call_tool("get_order", ARGS)) == {}
E AssertionError: assert {'code': 'RAT...trying.', ...} == {}
E
E Left contains 5 more items:
E {'code': 'RATE_LIMIT_EXCEEDED',
E 'details': {'retry_after_seconds': 0.999},
E 'message': 'Rate limit exceeded',
E 'retryable': True,
E 'suggestion': 'Wait 1 second before retrying.'}
E Use -v to get more diff
tests/test_middleware.py:16: AssertionError
------------------------------ Captured log call -------------------------------
WARNING promptise.server:_app.py:1067 MCPServer 'orders': RateLimitMiddleware runs before AuthMiddleware for tools ['get_order', 'queue_cancel', 'queue_list', 'queue_result', 'queue_status', 'queue_submit', 'refund_order']: no caller is identified yet when it picks a bucket, so every caller shares one 'global' bucket. Add AuthMiddleware first (middleware runs in the order it was added), or pass key_func= to RateLimitMiddleware if one shared bucket is intended.
=========================== short test summary info ============================
FAILED tests/test_middleware.py::test_each_caller_has_its_own_rate_limit - As...
1 failed, 1 passed in 0.15sSam was refused by a limit he never used. With the limiter in front of authentication, every caller is anonymous to it, so the whole server shares one bucket of five calls a burst. Since Promptise 1.3.3 the server notices and logs the warning you see under "Captured log call", once, at startup or on a TestClient's first call, and names every tool affected. A warning is easy to miss in a busy log, though, and nothing else fails: in production the symptom would be an agent refused because another one was busy. The test turns the order into a failing build.
The read-only test does the same for a custom middleware. ReadOnlyMode reads the tool's destructive_hint from ctx.state["tool_def"], so one switch freezes every destructive tool, including ones added later, while reads keep working.
[09]
Test MCP resources and prompts
Resources and prompts run through the same pipeline as tools: authentication, guards and middleware. (MCP Tools vs Resources vs Prompts explains when to use which.) One difference changes how you write the tests. call_tool returns an error envelope, but read_resource and get_prompt raise the error, which is what an MCP client receives for them, so you test them with pytest.raises:
import json
import pytest
from promptise.mcp.server import AuthenticationError, TestClient, ToolError
async def test_the_resource_inventory(agent):
resources = {str(r.uri) for r in await agent.list_resources()}
templates = {t.uriTemplate for t in await agent.list_resource_templates()}
assert resources == {"policy://refunds", "docs://manifest"} # the manifest is added for you
assert templates == {"orders://{order_id}/summary"}
async def test_read_a_static_resource(agent):
(item,) = await agent.read_resource_contents("policy://refunds")
assert item.mimeType == "text/plain"
assert "second supervisor" in item.text
async def test_read_a_resource_template(agent):
summary = json.loads(await agent.read_resource("orders://A-1001/summary"))
assert summary == {"order_id": "A-1001", "summary": "paid, 129.00 USD"}
async def test_resources_check_the_tenant_too(client_for):
globex = client_for("gina", tenant="globex")
with pytest.raises(ToolError, match="There is no order A-1001"):
await globex.read_resource("orders://A-1001/summary")
async def test_resources_need_a_token(app):
with pytest.raises(AuthenticationError):
await TestClient(app.server).read_resource("policy://refunds")
async def test_unknown_resources_raise_value_error(agent):
with pytest.raises(ValueError, match="Resource not found"):
await agent.read_resource("orders://A-1001/invoice")
async def test_prompt_arguments_arrive_as_strings_and_are_converted(agent):
result = await agent.get_prompt("refund_reply", {"order_id": "A-1001", "amount_cents": "1250"})
text = result.messages[0].content.text
assert text == "Write a short, friendly reply confirming a refund of 12.50 USD for order A-1001."
async def test_prompts_list_their_arguments(agent):
(prompt,) = await agent.list_prompts()
arguments = [(argument.name, argument.required) for argument in prompt.arguments]
assert arguments == [("order_id", True), ("amount_cents", True)]Three details the tests pin down:
The resource list includes `docs://manifest`. Promptise registers it on every server unless you pass auto_manifest=False, so an exact inventory has to name it.
An unknown URI raises `ValueError`, not an MCP error. That's TestClient's convention for "not found", for resources and prompts alike.
Prompt arguments arrive as strings. MCP clients send every prompt argument as a string; Promptise converts "1250" to the int the handler declares. The test sends the string, as a client would.
[10]
Test the background job queue
MCPQueue adds five tools to the server and runs jobs in worker tasks. Those workers start in a server startup hook, and TestClient doesn't run startup hooks. Start and stop them in an async fixture:
import asyncio
import json
import pytest
from helpers import error_of
RECONCILE = {"job_type": "reconcile", "args": {"day": "2026-10-10"}}
@pytest.fixture
async def queue(app):
"""TestClient doesn't run startup hooks, so start the workers yourself."""
await app.queue.start()
yield app.queue
await app.queue.stop()
async def finished(queue, job_id):
"""Wait through the Python API, which spends no caller's rate limit."""
for _ in range(250):
status = await queue.status(job_id)
if status["status"] not in ("pending", "running"):
return status
await asyncio.sleep(0.02)
raise TimeoutError(f"job {job_id} is still {status['status']}")
async def test_a_job_runs_to_completion(queue, agent):
job = json.loads((await agent.call_tool("queue_submit", RECONCILE))[0].text)
assert job["status"] == "pending"
assert (await finished(queue, job["job_id"]))["status"] == "completed"
result = await agent.call_tool("queue_result", {"job_id": job["job_id"]})
assert json.loads(result[0].text)["result"] == {"day": "2026-10-10", "matched": 412, "mismatched": 0}
async def test_bad_job_arguments_are_refused_at_submit(queue, agent):
wrong = {"job_type": "reconcile", "args": {"date": "2026-10-10"}}
error = error_of(await agent.call_tool("queue_submit", wrong))
assert error["code"] == "INVALID_JOB_ARGUMENTS"
async def test_callers_cannot_see_each_others_jobs(queue, client_for):
maya, sam = client_for("maya"), client_for("sam")
job = json.loads((await maya.call_tool("queue_submit", RECONCILE))[0].text)
lookup = {"job_id": job["job_id"]}
assert error_of(await sam.call_tool("queue_status", lookup))["code"] == "JOB_NOT_FOUND"
assert error_of(await maya.call_tool("queue_status", lookup)) == {}The first version of this file polled the job through the queue_status tool every 20 milliseconds. It failed with KeyError: 'status': on the sixth call, the rate limiter from the previous section answered RATE_LIMIT_EXCEEDED, and an error envelope has no status key. The test was right to fail. An agent polling a job spends its rate limit just the same, so poll slowly from agents, or raise the limit for the queue's tools. In tests, wait through the queue's Python API, which isn't a tool call and spends no one's rate limit.
The ownership test is the one to keep. A job belongs to the caller who submitted it, and another caller gets JOB_NOT_FOUND, exactly as for an ID that doesn't exist, so job IDs can't be probed. MCP Long-Running Tasks: Background Jobs in Python covers the queue itself in depth.
[11]
Add a few wire-level tests
Two promises live in the HTTP transport, where TestClient can't see them: a caller without a token is refused before any MCP message is processed, and a failed tool call carries isError: true. A module-scoped fixture runs the server's ASGI app on a loopback port for the whole file, and Promptise's MCPClient talks to it like any client:
"""What only the HTTP transport does. One server for the module, on a loopback port."""
import threading
import time
import pytest
import uvicorn
from helpers import SECRET
from orders_server import create_app
from promptise.mcp.client import MCPClient, MCPConnectionRejectedError
PORT = 9410 # any free port
@pytest.fixture(scope="module")
def url():
app = create_app(jwt_secret=SECRET)
config = uvicorn.Config(app.server.asgi_app(), host="127.0.0.1", port=PORT, log_level="warning")
server = uvicorn.Server(config)
thread = threading.Thread(target=server.run, daemon=True)
thread.start()
while not server.started:
time.sleep(0.01)
yield f"http://127.0.0.1:{PORT}/mcp"
server.should_exit = True
thread.join(timeout=5)
async def test_no_token_is_refused_before_mcp(url):
with pytest.raises(MCPConnectionRejectedError) as rejected:
async with MCPClient(url=url):
pass
assert rejected.value.status_code == 401
async def test_tool_errors_are_flagged_as_errors(url, token):
async with MCPClient(url=url, bearer_token=token("support-bot", roles=("agent",))) as client:
ok = await client.call_tool("get_order", {"order_id": "A-1001"})
denied = await client.call_tool("refund_order", {"order_id": "A-1001", "amount": 100})
assert ok.isError is False
assert denied.isError is True
assert '"code": "ACCESS_DENIED"' in denied.content[0].textMCPConnectionRejectedError carries the HTTP status, so the first test checks for a real 401, not just "some error". The second checks that the transport flags the refused refund as an error, which is how a client tells a failure from a result. Keep this layer thin. Everything else these two tests touch is already covered, faster, above.
[12]
Run the whole suite
pytest -v============================= test session starts ==============================
…
collected 35 items
tests/test_auth.py::test_no_token PASSED [ 2%]
tests/test_auth.py::test_bad_tokens_are_refused[expired] PASSED [ 5%]
tests/test_auth.py::test_bad_tokens_are_refused[forged] PASSED [ 8%]
tests/test_auth.py::test_bad_tokens_are_refused[other-audience] PASSED [ 11%]
tests/test_auth.py::test_headers_apply_to_one_call PASSED [ 14%]
tests/test_errors.py::test_error_envelopes[bad-order-id] PASSED [ 17%]
tests/test_errors.py::test_error_envelopes[zero-amount] PASSED [ 20%]
tests/test_errors.py::test_error_envelopes[too-large] PASSED [ 22%]
tests/test_errors.py::test_error_envelopes[unknown-order] PASSED [ 25%]
tests/test_errors.py::test_error_envelopes[unknown-tool] PASSED [ 28%]
tests/test_errors.py::test_details_tell_the_model_what_to_fix PASSED [ 31%]
tests/test_errors.py::test_a_mapped_exception_becomes_a_retryable_error PASSED [ 34%]
tests/test_errors.py::test_an_unexpected_exception_does_not_leak PASSED [ 37%]
tests/test_middleware.py::test_each_caller_has_its_own_rate_limit PASSED [ 40%]
tests/test_middleware.py::test_read_only_mode_blocks_writes_but_not_reads PASSED [ 42%]
tests/test_queue.py::test_a_job_runs_to_completion PASSED [ 45%]
tests/test_queue.py::test_bad_job_arguments_are_refused_at_submit PASSED [ 48%]
tests/test_queue.py::test_callers_cannot_see_each_others_jobs PASSED [ 51%]
tests/test_resources_prompts.py::test_the_resource_inventory PASSED [ 54%]
tests/test_resources_prompts.py::test_read_a_static_resource PASSED [ 57%]
tests/test_resources_prompts.py::test_read_a_resource_template PASSED [ 60%]
tests/test_resources_prompts.py::test_resources_check_the_tenant_too PASSED [ 62%]
tests/test_resources_prompts.py::test_resources_need_a_token PASSED [ 65%]
tests/test_resources_prompts.py::test_unknown_resources_raise_value_error PASSED [ 68%]
tests/test_resources_prompts.py::test_prompt_arguments_arrive_as_strings_and_are_converted PASSED [ 71%]
tests/test_resources_prompts.py::test_prompts_list_their_arguments PASSED [ 74%]
tests/test_tools.py::test_the_tool_inventory PASSED [ 77%]
tests/test_tools.py::test_the_model_sees_described_inputs PASSED [ 80%]
tests/test_tools.py::test_get_order PASSED [ 82%]
tests/test_tools.py::test_a_tenant_cannot_read_another_tenants_order PASSED [ 85%]
tests/test_tools.py::test_a_supervisor_can_refund PASSED [ 88%]
tests/test_tools.py::test_an_agent_cannot_refund PASSED [ 91%]
tests/test_tools.py::test_an_order_is_refunded_only_once PASSED [ 94%]
tests/test_wire.py::test_no_token_is_refused_before_mcp PASSED [ 97%]
tests/test_wire.py::test_tool_errors_are_flagged_as_errors PASSED [100%]
============================== 35 passed in 1.30s ==============================That's the suite on Python 3.12 with promptise 1.3.3, mcp 1.30.0, pytest 9.1.1 and pytest-asyncio 1.4.0. It ran thirteen times while this guide was written, and all 35 tests passed every time, in 0.87 to 2.30 seconds, with no warnings.
[13]
Run the tests in CI with GitHub Actions
The workflow installs the project with its test extras and runs pytest on every push and pull request:
name: tests
on:
push:
pull_request:
jobs:
pytest:
runs-on: ubuntu-latest
timeout-minutes: 10
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install
run: pip install -e ".[test]"
- name: Test
run: pytest -qThe tests need no secrets, no model and no services. The only network use is the wire fixture's loopback port. This workflow wasn't run on GitHub, and act wasn't available locally, so its two steps were run as written in a fresh virtual environment, from a clean copy of the project:
pip install -e ".[test]"
pytest -q................................... [100%]
35 passed in 2.08s[14]
Honest limits
These are true of Promptise Foundry 1.3.3:
`TestClient` is not the transport. No HTTP auth gate, no isError flag, no Host or Origin checks, no CORS and no MCP sessions. A call without a token gets an AUTHENTICATION_ERROR envelope from AuthMiddleware, where the live server answers 401. And list_tools() without a token listed all seven tools of this require_auth=True server, which over HTTP would never get that far. Cover those with a few wire-level tests.
Startup and shutdown hooks don't run. Without queue.start(), a submitted job was still pending half a second later. The same goes for your own @server.on_startup hooks, such as opening a database pool: open it in a fixture.
Errors arrive in two shapes. Tool calls return the envelope as JSON text; resource reads and prompt requests raise. An unknown resource or prompt raises ValueError.
There is no `dependency_overrides`. Pass fakes into a server factory, as create_app(payments=...) does here.
Declared rate limits are per client object. A limit declared with @server.tool(rate_limit="2/min") keeps its buckets in each TestClient. In a check, two anonymous TestClients on one server got two separate buckets, where the live server would give them one. A RateLimitMiddleware added to the server is shared, as in this guide.
Custom guards don't imply authentication. The built-in identity guards and roles= do. A guard you write yourself needs auth=True on the tool or require_auth=True on the server, or it sees an anonymous caller: in a check, a custom guard on a tool without auth=True got client_id None for a valid token and denied it.
The wire fixture uses a fixed port. Two test processes on one machine, for example with pytest-xdist, need a port each.
None of this tests the model. These tests prove what the server returns for a call. Whether a model makes the right call is a different test: see How to Test LLM Prompts and AI Agents in Python.
[15]
Frequently asked questions
How do you unit test an MCP server in Python?
Call the server in-process instead of over a transport. With Promptise Foundry, wrap the MCPServer in TestClient(server, meta={...}) and use call_tool, read_resource and get_prompt in async pytest tests. Each call runs validation, authentication, guards, middleware and your handler, and returns what a client would receive. Build the server in a factory function so each test gets fresh state and fakes for its dependencies.
Do I need pytest-asyncio to test MCP servers?
You need something that runs async def tests, because every TestClient method is a coroutine. pytest-asyncio with asyncio_mode = "auto" is the least ceremony: no marker on each test or fixture. In its default strict mode, mark each test with @pytest.mark.asyncio instead.
How do you test MCP authentication without an identity provider?
Use a test-only signing secret. JWTAuth.create_token mints tokens with whatever subject, tenant and roles a test needs, and APIKeyAuth takes keys straight from a dict. Test the refusals as well as the happy path: no token, an expired token, a forged signature and a token minted for a different audience. JwksAuth fetches public keys over HTTP, so testing it needs a reachable JWKS URL; AI Agent Identity: Authenticate Agents to MCP Servers runs a small local issuer for exactly that.
Is TestClient the same as the MCP Inspector?
No. The MCP Inspector is an interactive tool: you connect it to a server and call tools by hand. TestClient is for automated tests: it needs no running server, and it's fast enough to run 35 checks in a second on every commit. Use the Inspector to explore, and tests to keep things from breaking.
Can I test an MCP server built with another SDK?
TestClient works with a Promptise MCPServer only. For any other MCP server, start it and drive it with an MCP client, as the wire tests above do with MCPClient, which speaks to any server over Streamable HTTP, SSE or stdio.
[16]
Where to go next
Testing: every TestClient method, with more examples.
Authentication & Security: providers, guards and ClientContext on the server.
Routers & Middleware: every built-in middleware and how to write your own.
Queue & Background Jobs: job types, progress, cancellation and ownership.
Build a Production MCP Server in Python (Step by Step): the server-side features these tests exercise, built one at a time.
Multi-Tenant AI Agents: Keep Each Customer's Data Apart: tenant isolation from the token to the tool, with its own tests.
MCP Transports: stdio vs Streamable HTTP vs SSE (Python): the layer the wire-level tests cover.