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
)

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

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.*"))
)

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(...)).

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="7.5.0")
# ...
server.stop()

API reference

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