Testing MCP Servers: Contract Tests, Mocks & Evals

Three distinct test layers catch three distinct classes of MCP server failure: contract tests catch schema regressions, mocked integration tests catch business-logic bugs quickly and deterministically, and model-in-the-loop evals catch tool-design problems that only show up when a real model tries to use the tool. Getting the CI cadence right for each layer matters as much as writing the tests themselves.

Layer 1: Contract Tests

Assert the exact shape of each tool’s schema so any unversioned change fails the build immediately, this is the automated enforcement of the versioning discipline from Lesson 16.

import pytest
from mcp.client.session import ClientSession

@pytest.mark.asyncio
async def test_kb_query_v2_contract(mcp_test_session: ClientSession):
    tools = await mcp_test_session.list_tools()
    tool = next(t for t in tools if t.name == "kb_query/v2")

    schema = tool.inputSchema
    assert set(schema["properties"].keys()) == {"query", "top_k", "doc_type", "min_relevance"}
    assert schema["properties"]["query"]["type"] == "string"
    assert "query" in schema.get("required", [])
    assert "doc_type" not in schema.get("required", [])  # Must stay optional, per Lesson 16

Layer 2: Mocked Integration Tests, Including Failure Injection

Test each tool’s actual logic against a mocked downstream system, fast, deterministic, and runnable in CI without live credentials or network access to production systems. It is not enough to only mock the happy path, the error-handling branches built in Lesson 6 and Lesson 13 (timeouts, 404s, 429s, partial failures) need their own tests, or they silently rot the first time a real downstream outage exercises a code path nothing has run since it was written.

@pytest.mark.asyncio
async def test_search_accounts_filters_by_tier(mock_crm_client):
    mock_crm_client.search.return_value = [
        Account(id="A1", name="Acme", tier="enterprise"),
        Account(id="A2", name="Beta", tier="startup"),
    ]

    result = await search_accounts(SearchAccountsInput(query="a", tier="enterprise"))

    mock_crm_client.search.assert_called_once_with("a", tier="enterprise", min_arr=0)
    assert result["count"] == 2  # Mock returns both; asserts the tool passes the filter through correctly

@pytest.mark.asyncio
async def test_query_orders_rejects_sql_injection_attempt(mock_db_pool):
    """Regression test for Lesson 14's parameterized-query requirement."""
    malicious_input = OrdersQueryInput(customer_id="1; DROP TABLE orders;--", limit=5)
    await query_orders(malicious_input)

    # Assert the value was passed as a bound parameter, never concatenated into query text
    call_args = mock_db_pool.fetch.call_args
    assert "DROP TABLE" not in call_args[0][0]  # Query text itself must never contain the payload
    assert call_args[0][1] == "1; DROP TABLE orders;--"  # Passed safely as a parameter value

@pytest.mark.asyncio
async def test_get_account_handles_downstream_timeout(mock_crm_client):
    """Failure-injection test: simulate a CRM timeout and assert the tool
    returns the retryable structured error from Lesson 6, not an unhandled
    exception that would surface as a confusing 500 to the caller."""
    mock_crm_client.get_account.side_effect = CRMTimeoutError()

    result = await get_account(account_id="A1042")

    assert result["isError"] is True
    assert result["retryable"] is True

@pytest.mark.asyncio
async def test_bulk_update_tier_reports_partial_failure(mock_crm_client):
    """Failure-injection test for Lesson 6's partial-success pattern:
    one account succeeds, one raises AccountNotFoundError."""
    async def side_effect(account_id, tier):
        if account_id == "A404":
            raise AccountNotFoundError()
    mock_crm_client.update_tier.side_effect = side_effect

    result = await bulk_update_tier(BulkUpdateInput(account_ids=["A1", "A404"], new_tier="enterprise"))

    assert result["succeeded"] == ["A1"]
    assert result["failed"] == [{"account_id": "A404", "reason": "not_found"}]

Failure-injection tests like these two are what actually validate the error-taxonomy design from Lesson 6, without them, a refactor that accidentally turns a caught CRMTimeoutError into an unhandled exception would pass every happy-path test and only surface in production during the next real timeout.

Layer 3: Model-in-the-Loop Evals

Run a fixed set of representative tasks against a real (or representative open-weight) model connected to the actual server, and score tool selection and argument correctness, not just server correctness, following the same methodology built out in Lesson 23.

EVAL_CASES = [
    {
        "task": "Find all high-priority support tickets for account A1042 from the last week.",
        "expected_tool": "support.search_tickets",
        "expected_args_contains": {"account_id": "A1042", "priority": "high"},
    },
    {
        "task": "What is Acme Corp's current subscription tier?",
        "expected_tool": "crm.search_accounts",  # Not support.search_tickets, tests disambiguation
        "expected_args_contains": {"query": "Acme"},
    },
]

async def run_tool_selection_eval(harness, eval_cases: list[dict]) -> dict:
    results = []
    for case in eval_cases:
        tool_call = await harness.run_single_turn(case["task"])
        correct_tool = tool_call.name == case["expected_tool"]
        correct_args = all(
            tool_call.arguments.get(k) == v for k, v in case["expected_args_contains"].items()
        )
        results.append({"task": case["task"], "correct_tool": correct_tool, "correct_args": correct_args})
    return {
        "tool_selection_accuracy": sum(r["correct_tool"] for r in results) / len(results),
        "argument_accuracy": sum(r["correct_args"] for r in results) / len(results),
    }

flowchart TD
    Layer1["Layer 1: Contract Tests\n(schema shape)"]
    Layer2["Layer 2: Mocked Integration Tests\n(business logic + failure injection)"]
    Layer3["Layer 3: Model-in-the-Loop Evals\n(tool selection quality)"]

    Layer1 -->|"Cheap, deterministic,\nrun on EVERY pull request"| Cadence1["CI: every PR"]
    Layer2 -->|"Fast, no live credentials,\nrun on EVERY pull request"| Cadence2["CI: every PR"]
    Layer3 -->|"Real inference calls,\nslower and costlier"| Cadence3["CI: only when tool name,\ndescription, or schema changes"]

    style Layer1 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style Layer2 fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style Layer3 fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style Cadence1 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style Cadence2 fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style Cadence3 fill:#fff7ed,stroke:#f59e0b,color:#0F172A

Run Layer 1 and 2 on every pull request unconditionally, they are fast and deterministic enough that there is no good reason not to. Layer 3 is meaningfully more expensive (real inference calls against a model, whether a hosted API or a self-hosted endpoint), so gate it specifically on changes to what actually affects tool-selection behavior, a tool’s name, its description, or its schema, per Lesson 22’s point that these are exactly the properties a model reasons over. A pull request that only refactors internal implementation without touching any of those three surfaces gains nothing from re-running the eval suite and should not pay its cost.

The next lesson handles a case none of these three layers directly test: tools whose execution genuinely takes longer than a reasonable synchronous request-response window.