MockServer Python Client

Python client for MockServer with full WebSocket callback support.

Features

• Full REST API: Create expectations, verify requests, clear/reset, retrieve recorded data
• Response Callbacks: Register Python functions that dynamically generate responses via WebSocket
• Forward Callbacks: Modify requests before they are forwarded to the real server
• Forward+Response Callbacks: Modify both the forwarded request and the response
• Fluent API: client.when(request).respond(callback) — mirrors the Java client
• Async + Sync: Native asyncio API with a synchronous wrapper for non-async code
• Minimal Dependencies: Only websockets (for callback support)

Installation

pip install mockserver-client

Quick Start

Synchronous API

from mockserver import MockServerClient, HttpRequest, HttpResponse

client = MockServerClient("localhost", 1080)

# Static expectation
client.when(
    HttpRequest.request("/api/users").with_method("GET")
).respond(
    HttpResponse.response('{"users": []}', status_code=200)
)

# Verify
client.verify(
    HttpRequest.request("/api/users").with_method("GET"),
    VerificationTimes.at_least(1)
)

# Clean up
client.reset()
client.close()

Context Manager

with MockServerClient("localhost", 1080) as client:
    client.when(
        HttpRequest.request("/hello")
    ).respond(
        HttpResponse.response("world")
    )

Async API

import asyncio
from mockserver import AsyncMockServerClient, HttpRequest, HttpResponse

async def main():
    async with AsyncMockServerClient("localhost", 1080) as client:
        await client.when(
            HttpRequest.request("/api/data")
        ).respond(
            HttpResponse.response('{"key": "value"}')
        )

asyncio.run(main())

Response Callbacks

Register a Python function that generates responses dynamically when matching requests arrive:

from mockserver import MockServerClient, HttpRequest, HttpResponse

def handle_request(request):
    if request.method == "POST":
        return HttpResponse.response("created", status_code=201)
    return HttpResponse.not_found_response()

client = MockServerClient("localhost", 1080)
client.mock_with_callback(
    HttpRequest.request("/api/callback"),
    handle_request
)

Or with the fluent API:

client.when(
    HttpRequest.request("/api/callback")
).respond(handle_request)

Forward Callbacks

Modify requests before they are forwarded to the real server:

def modify_request(request):
    return request.with_header("X-Forwarded", "true").with_path("/modified" + request.path)

client.mock_with_forward_callback(
    HttpRequest.request("/proxy/.*"),
    modify_request
)

Forward+Response Callbacks

Modify both the forwarded request and the response:

def modify_request(request):
    return request.with_header("X-Proxied", "true")

def modify_response(request, response):
    return response.with_header("X-Modified", "true")

client.mock_with_forward_callback(
    HttpRequest.request("/proxy/.*"),
    modify_request,
    modify_response
)

Forward with Overrides

Forward the matched request with fields replaced (request_override), modified (request_modifier), or with its response overridden (response_override, response_modifier, response_template), without a callback:

from mockserver import HttpOverrideForwardedRequest, HttpRequest, HttpResponse

client.when(HttpRequest.request("/some/path")).forward(
    HttpOverrideForwardedRequest(
        request_override=HttpRequest(path="/other/path", headers={"Host": ["target.host.com"]}),
        response_override=HttpResponse(headers={"X-Proxied": ["true"]}),
        response_modifier={"headers": {"remove": ["Server"]}},
    )
)

http_request and http_response are deprecated aliases of request_override and response_override, kept for existing code; set one name of each pair, not both.

Verification

from mockserver import VerificationTimes

# Verify a request was received at least once
client.verify(
    HttpRequest.request("/api/users").with_method("GET"),
    VerificationTimes.at_least(1)
)

# Verify exact count
client.verify(
    HttpRequest.request("/api/users"),
    VerificationTimes.exactly(3)
)

# Verify request sequence (order matters)
client.verify_sequence(
    HttpRequest.request("/first"),
    HttpRequest.request("/second"),
    HttpRequest.request("/third"),
)

# Verify no interactions
client.verify_zero_interactions()

Request Matching

Headers, query parameters and cookies

Pass headers, query and path parameters, trailers and cookies as a dict, as in MockServer's JSON, or as a list of KeyToMultiValue. A "keyMatchStyle" key (SUB_SET, the default, or MATCHING_KEY) sets how the values of a key are matched.

from mockserver import HttpRequest, HttpResponse

client.when(
    HttpRequest(
        path="/search",
        headers={"Accept": "application/json"},        # one value
        query_string_parameters={"tag": ["a", "b"]},   # several values
        cookies={"session": "abc"},
    )
).respond(
    HttpResponse(status_code=302, headers={"Location": ["https://example.com"]})
)

# every value the request has for X-Tag must match, not just one
HttpRequest(headers={"keyMatchStyle": "MATCHING_KEY", "X-Tag": ["a", "b"]})

JWT matcher

Match requests by the claims (and metadata) of a JWT carried in a header. Each claim value is exact-or-regex; prefix a value with ! to negate it.

from mockserver import HttpRequest, HttpResponse, Jwt

client.when(
    HttpRequest.request("/secure")
        .with_method("GET")
        .with_jwt(
            claims={
                "sub": "user-123",           # exact match
                "email": "^.+@example.com$", # regex match
                "role": "!admin",            # negated — matches any role except admin
            },
            issuer="https://issuer.example.com",
            audience="my-api",
            algorithm="RS256",
            header="authorization",          # optional (defaults to Authorization)
            scheme="Bearer",                 # optional token scheme
        )
).respond(
    HttpResponse.response("granted", status_code=200)
)

# ...or pass a Jwt object directly:
HttpRequest.request("/secure").with_jwt(Jwt(claims={"sub": "user-123"}))

allOf body matcher

Require a request body to satisfy all of several body matchers at once.

from mockserver import AllOfBody, Body, JsonPathBody, RegexBody, HttpRequest

client.when(
    HttpRequest.request("/api/orders")
        .with_method("POST")
        .with_body(AllOfBody(body_all_of=[
            JsonPathBody(json_path="$.name"),   # body has a name field
            RegexBody(regex=".*active.*"),      # and mentions "active"
        ]))
).respond(
    HttpResponse.response("ok")
)

# Shorthand factory:
HttpRequest.request("/api/orders").with_body(
    Body.all_of(JsonPathBody(json_path="$.name"), RegexBody(regex=".*active.*"))
)

Conditional (if/then/else) matcher

Pass a ConditionalRequestDefinition where a request matcher goes: when if_request matches, then_request must match too; otherwise else_request must (with no else_request it matches whenever if_request does not). Each branch is an HttpRequest, an OpenAPIDefinition or another ConditionalRequestDefinition. Retrieved expectations whose httpRequest holds if read back as this type.

from mockserver import Body, ConditionalRequestDefinition, HttpRequest, HttpResponse

client.when(
    ConditionalRequestDefinition(
        if_request=HttpRequest(method="POST").with_header("content-type", "application/json"),
        then_request=HttpRequest(body=Body.json_schema('{"type": "object", "required": ["orderId"]}')),
        else_request=HttpRequest(method="GET"),
    )
).respond(HttpResponse(status_code=200))

Retrieval

# Get recorded requests
requests = client.retrieve_recorded_requests(
    HttpRequest.request("/api/.*")
)

# Get active expectations
expectations = client.retrieve_active_expectations()

# Get log messages
logs = client.retrieve_log_messages()

Control

# Clear specific expectations
client.clear(HttpRequest.request("/api/users"))

# Clear by type
client.clear(HttpRequest.request("/api/users"), clear_type="LOG")

# Reset everything
client.reset()

# Bind additional ports
client.bind(1081, 1082)

# Check if running
if client.has_started():
    print("MockServer is running")

# Stop
client.stop()

AI Protocol Mocking

Declarative builders mock an MCP (Model Context Protocol) server or an A2A (Agent-to-Agent) server with a single fluent chain. Each builder produces a set of HTTP expectations that speak JSON-RPC 2.0 and echo the incoming request id.

from mockserver import mcp_mock, a2a_mock

# Mock an MCP server (Streamable HTTP, JSON-RPC 2.0) on /mcp
mcp_mock() \
    .with_tool("get_weather") \
        .with_description("Get weather for a city") \
        .with_input_schema('{"type": "object", "properties": {"city": {"type": "string"}}}') \
        .responding_with("72F and sunny") \
        .and_() \
    .apply_to(client)

# Mock an A2A agent on /a2a (agent card + tasks/send|get|cancel)
a2a_mock() \
    .with_agent_name("TranslatorAgent") \
    .with_skill("translate") \
        .with_name("Translation") \
        .with_description("Translates text between languages") \
        .with_tag("i18n") \
        .with_example("Translate hello to Spanish") \
        .and_() \
    .on_task_send() \
        .matching_message("translate.*") \
        .responding_with("Hola") \
        .and_() \
    .apply_to(client)

The A2A builder also supports streaming (SSE) and push notifications:

a2a_mock() \
    .with_streaming() \
    .with_push_notifications("http://localhost:1234/callback") \
    .apply_to(client)

build() returns the list of Expectation objects without registering them, so you can inspect or persist them; apply_to(client) registers them via upsert.

SRE / Resilience

Verify a service-level objective over recorded SLI samples, or run a scheduled multi-stage chaos experiment. Both require the corresponding server feature to be enabled (sloTrackingEnabled, chaos experiments).

# Verify an SLO — a FAIL verdict (HTTP 406) raises MockServerVerificationError
verdict = client.verify_slo({
    "name": "checkout",
    "minimumSampleCount": 100,
    "objectives": [
        {"sli": "errorRate", "comparator": "LESS_THAN", "threshold": 0.01},
        {"sli": "p99LatencyMs", "comparator": "LESS_THAN", "threshold": 250},
    ],
})
print(verdict["result"])  # PASS or INCONCLUSIVE

# Start a multi-stage chaos experiment (only one may be active at a time)
client.start_chaos_experiment({
    "name": "latency-injection",
    "loop": False,
    "stages": [
        {"durationMillis": 60000, "profiles": {"payments.svc": {"latencyMs": 500}}},
    ],
})

Both methods are available on the async client too (await client.verify_slo(...), await client.start_chaos_experiment(...)).

Diagnose Matching

explain(correlation_id) tells you which expectation would serve a recorded request now and why each expectation ahead of it was passed over; debug_mismatch(request) compares a request with every active expectation; explain_unmatched(limit=..., suggest_fixes=...) explains recent requests that matched nothing. Each returns a typed dataclass (RequestExplanation, DebugMismatchResult, ExplainUnmatchedResult) and raises MockServerError with the server's reason on failure. The async client has the same methods.

explanation = client.explain("abc-123-def-456")
for skipped in explanation.skipped_expectations:
    print(skipped.expectation_id, skipped.reason, skipped.explanation)

TLS Support

# Uses system trust store (default — verifies certificates)
client = MockServerClient("localhost", 1080, secure=True)

# Custom CA certificate
client = MockServerClient(
    "localhost", 1080,
    secure=True,
    ca_cert_path="/path/to/ca.pem"
)

# Disable certificate verification (testing only — NOT recommended for production)
client = MockServerClient(
    "localhost", 1080,
    secure=True,
    tls_verify=False
)

Domain Model

All domain model classes support builder-style chaining:

request = (
    HttpRequest.request("/api/users")
    .with_method("POST")
    .with_header("Content-Type", "application/json")
    .with_header("Authorization", "Bearer token")
    .with_body('{"name": "test"}')
    .with_query_param("page", "1")
    .with_secure(True)
)

response = (
    HttpResponse.response()
    .with_status_code(201)
    .with_header("Location", "/api/users/1")
    .with_body('{"id": 1, "name": "test"}')
    .with_delay(Delay(time_unit="SECONDS", value=1))
)

Interactive Breakpoints

The client supports matcher-driven interactive breakpoints over the callback WebSocket. Register a breakpoint matcher to pause forwarded/proxied exchanges at specific phases and inspect/modify/continue them via callback handlers.

Register a breakpoint (sync client)

from mockserver import MockServerClient, HttpRequest, HttpResponse

client = MockServerClient("localhost", 1080)

# REQUEST phase only
bp_id = client.add_request_breakpoint(
    HttpRequest(path="/api/.*"),
    lambda request: request,  # continue unchanged (or return HttpResponse to abort)
)

# REQUEST + RESPONSE
bp_id = client.add_request_and_response_breakpoint(
    HttpRequest(path="/api/.*"),
    lambda request: request,                      # REQUEST handler
    lambda request, response: response,           # RESPONSE handler
)

# All phases with stream frame handler
bp_id = client.add_breakpoint(
    HttpRequest(path="/stream/.*"),
    ["REQUEST", "RESPONSE", "RESPONSE_STREAM", "INBOUND_STREAM"],
    request_handler=lambda request: request,
    response_handler=lambda request, response: response,
    stream_frame_handler=lambda frame: {"action": "CONTINUE"},
    # Other actions: MODIFY (with body), DROP, INJECT (with body), CLOSE
)

Manage breakpoints

# List all matchers
matchers = client.list_breakpoint_matchers()  # {"matchers": [...]}

# Remove a specific matcher
client.remove_breakpoint_matcher(bp_id)

# Clear all matchers
client.clear_breakpoint_matchers()

The async client (AsyncMockServerClient) exposes the same methods as coroutines.

Start / Launch MockServer

The Python client can download and launch a local MockServer instance directly -- no Java installation and no Docker required. The launcher downloads a self-contained platform bundle (mockserver-<version>-<os>-<arch>) from the GitHub Release, verifies its SHA-256, caches it per-user, and starts it.

Quick start

from mockserver.launcher import start, MockServerProcess

# Download (first run) and start MockServer on port 1080
with start(port=1080) as server:
    print(f"MockServer running on port {server.port}, PID {server.pid}")
    # ... use MockServer ...
# Server is stopped automatically when the context manager exits

Just ensure the binary is present

from mockserver.launcher import ensure_binary

launcher_path = ensure_binary()  # returns Path to the launcher executable

Specify a version

from mockserver.launcher import start

server = start(port=1080, version="9.0.0")
# ...
server.stop()

API reference

On Windows the launcher runs mockserver.bat through cmd.exe with every argument quoted, so characters such as &, | and ^ reach MockServer unchanged. start() raises ValueError if the launcher path or an extra_args entry contains a double quote, a line break or NUL, or if together they contain more than one % (which cmd.exe would expand as a %NAME% variable). No shell is used on Linux or macOS.

Supported platforms

Environment variables

Version

By default the launcher downloads the MockServer version matching this client package (currently the version set in pyproject.toml). Pass an explicit version argument to override.

Requirements

• Python 3.9+
• websockets >= 12.0 (for callback support)

License

Apache 2.0

AI Assistant Integration

MockServer includes a built-in MCP (Model Context Protocol) server that enables AI coding assistants to create expectations, verify requests, and debug HTTP traffic programmatically.

• MCP Endpoint: http://localhost:1080/mockserver/mcp
• AI Documentation: llms.txt
• Setup Guide: AI Integration