Serve MCP over stdio and Streamable HTTP, and test the contracts

A transport decides who starts your MCP server, how many clients share it, and what can go wrong on the wire. We serve the typed task-tracker server from the previous lesson over both standard transports: stdio, where each host starts its own server process, and Streamable HTTP, where one long-running process answers every client. You read the JSON-RPC and HTTP headers the SDK sends, then pin the server’s contract with pytest: fast in-process tests for the schemas and every call, and one integration test for each transport.

When you need this. The task tracker has to work in two places: inside one developer’s editor and as a shared service that the whole team’s agents call. The editor host can start the server as a child process over stdio; the shared service needs Streamable HTTP, one process behind a URL. The transport decides who can connect and where state lives, so we settle it before we write the tests.

1. Set up the project

Download the complete transport lab and extract it with its fixtures, test files and README. Keep the typed-tools lab extracted beside mcp-transports/: the tests compare their server files. The only addition to the previous lesson is pytest; the async tests use the pytest plugin that ships with AnyIO, which the SDK already installs.

unzip -n implement-an-mcp-server-with-typed-tool-contracts.zip
unzip serve-mcp-over-stdio-and-streamable-http-and-test-the-contracts.zip
cd mcp-transports
python3.12 -m venv .venv
.venv/bin/python -m pip install -r requirements-lock.txt

2. Choose a transport

Both transports carry the same JSON-RPC messages. They differ in process ownership, and every other difference follows from that:

Choose stdio when the server acts for one user on one machine: the operating system’s process boundary is the access control, and there is no port to protect. Choose Streamable HTTP when several clients must share one server, or when the server runs somewhere else. That choice moves authentication, concurrency and state isolation into your code, as the production section below explains. The SDK also has an "sse" transport; the specification deprecates HTTP+SSE in favor of Streamable HTTP, so new servers should not use it.

3. Serve one server on both transports

Now we put the server on the wire. serve.py imports the unchanged server and picks a transport from the command line. MCPServer.run() in mcp 2.3.0 accepts "stdio", "sse" or "streamable-http"; with "streamable-http" it starts uvicorn and serves the MCP endpoint at /mcp. To mount the server inside an existing ASGI application instead, server.streamable_http_app() returns the Starlette app.

"""Serve the tasks server over stdio or Streamable HTTP.

Run: python serve.py                      (stdio: the host starts this process)
     python serve.py --http --port 8000   (Streamable HTTP at http://127.0.0.1:8000/mcp)
"""
import argparse

from tasks_server import server

parser = argparse.ArgumentParser()
parser.add_argument("--http", action="store_true", help="serve Streamable HTTP instead of stdio")
parser.add_argument("--port", type=int, default=8000)
args = parser.parse_args()

if args.http:
    server.run("streamable-http", host="127.0.0.1", port=args.port)
else:
    server.run()

Bind to 127.0.0.1, not 0.0.0.0. The specification says local servers should bind only to localhost, and the SDK switches on its Host and Origin allowlist only for localhost addresses.

client.py takes either stdio or a URL; the SDK Client treats a URL string as Streamable HTTP. Start the HTTP server in one terminal, then run the client three times:

.venv/bin/python serve.py --http --port 8000

# in a second terminal
.venv/bin/python client.py stdio
.venv/bin/python client.py http://127.0.0.1:8000/mcp
.venv/bin/python client.py http://127.0.0.1:8000/mcp

server   tasks 1.0.0 protocol 2026-07-28
tools    ['list_tasks', 'add_task', 'complete_task']
added    T-4 open count 3
server   tasks 1.0.0 protocol 2026-07-28
tools    ['list_tasks', 'add_task', 'complete_task']
added    T-4 open count 3
server   tasks 1.0.0 protocol 2026-07-28
tools    ['list_tasks', 'add_task', 'complete_task']
added    T-5 open count 4

"""Connect to the tasks server over one transport and make two calls.

Run: python client.py stdio                       (starts serve.py as a subprocess)
     python client.py http://127.0.0.1:8000/mcp   (needs: python serve.py --http)
"""
import asyncio
import sys
from pathlib import Path

from mcp import Client, StdioServerParameters


def target(arg: str):
    if arg == "stdio":
        return StdioServerParameters(command=sys.executable, args=[str(Path(__file__).with_name("serve.py"))])
    return arg  # a URL string selects Streamable HTTP


async def main(arg: str) -> None:
    async with Client(target(arg)) as client:
        print("server  ", client.server_info.name, client.server_info.version, "protocol", client.protocol_version)
        print("tools   ", [tool.name for tool in (await client.list_tools()).tools])
        added = await client.call_tool("add_task", {"title": "Tag v1.2"})
        listed = await client.call_tool("list_tasks", {"status": "open"})
        print("added   ", added.structured_content["id"], "open count", listed.structured_content["count"])


if __name__ == "__main__":
    asyncio.run(main(sys.argv[1]))

The three runs added T-4, T-4, T-5. The stdio client started its own server process, so it saw the three seed tasks. The second HTTP run reached the same process as the first, so it saw the task the first run had added. That is the state model of the table above, recorded.

A print() in a stdio server. The recorded client logs Failed to parse JSONRPC message from server and skips the line, although the tool calls still work. Reserve stdout for protocol messages and send application logs to stderr with logging.

Wrong endpoint path. The client receives HTTP 404 and raises MCPError with Not Found. Use /mcp, or set the matching streamable_http_path on the server.

State shared between HTTP clients. The second recorded HTTP client sees the first client’s task. Store state per authenticated caller, or choose stdio when each user needs a separate server process.

4. Pin the contract with in-process tests

With both transports working, we can pin what clients rely on. Clients depend on the published schemas, not on your Python code. A renamed field or a loosened pattern breaks them without failing any unit test of the function. So snapshot the tools/list result into a file, review its diff like code, and test the live server against it. contract.py writes the snapshot, tools-contract.json, with 3 tool definitions.

This is the add_task entry. If a change raises minLength here, calls that passed before now fail, and the diff shows it in review:

{
  "name": "add_task",
  "description": "Add an open task and return it with its new ID.",
  "inputSchema": {
    "type": "object",
    "properties": {
      "title": {
        "maxLength": 80,
        "minLength": 3,
        "title": "Title",
        "type": "string"
      },
      "priority": {
        "default": "normal",
        "enum": [
          "low",
          "normal",
          "high"
        ],
        "title": "Priority",
        "type": "string"
      }
    },
    "required": [
      "title"
    ],
    "title": "add_taskArguments"
  },
  "outputSchema": {
    "description": "One task in the tracker.",
    "properties": {
      "id": {
        "description": "Task ID such as T-1.",
        "title": "Id",
        "type": "string"
      },
      "title": {
        "title": "Title",
        "type": "string"
      },
      "priority": {
        "enum": [
          "low",
          "normal",
          "high"
        ],
        "title": "Priority",
        "type": "string"
      },
      "status": {
        "enum": [
          "open",
          "done"
        ],
        "title": "Status",
        "type": "string"
      }
    },
    "required": [
      "id",
      "title",
      "priority",
      "status"
    ],
    "title": "Task",
    "type": "object"
  }
}

Contract tests pass the MCPServer object straight to Client. The in-process connection runs the full request handling, with argument validation, tool code and output conversion, but without a subprocess or a socket, so each test takes milliseconds. The fixture reloads the server module so that every test starts from the same three tasks.

"""Contract tests: an in-process client checks the published tools and every call. Run: python -m pytest -v"""
import importlib
import json
from pathlib import Path

import pytest
from jsonschema import validate
from mcp import Client

import contract
import tasks_server

pytestmark = pytest.mark.anyio
PUBLISHED = {tool["name"]: tool for tool in json.loads(contract.CONTRACT.read_text(encoding="utf-8"))}


@pytest.fixture
def anyio_backend():
    return "asyncio"


@pytest.fixture
async def client():
    server = importlib.reload(tasks_server).server  # a fresh task list for every test
    async with Client(server) as client:
        yield client


async def test_tools_list_matches_the_published_contract(client):
    tools = (await client.list_tools()).tools
    assert [tool.model_dump(mode="json", by_alias=True, exclude_none=True) for tool in tools] == list(PUBLISHED.values())


@pytest.mark.parametrize("name, arguments", [
    ("list_tasks", {}),
    ("list_tasks", {"status": "done"}),
    ("add_task", {"title": "Tag v1.2", "priority": "high"}),
    ("complete_task", {"task_id": "T-1"}),
])
async def test_each_tool_returns_content_that_matches_its_output_schema(client, name, arguments):
    result = await client.call_tool(name, arguments)
    assert not result.is_error
    validate(result.structured_content, PUBLISHED[name]["outputSchema"])
    assert json.loads(result.content[0].text) == result.structured_content


@pytest.mark.parametrize("name, arguments, message", [
    ("add_task", {"title": "v2"}, "String should have at least 3 characters"),
    ("add_task", {"title": "Tag v1.2", "priority": "urgent"}, "Input should be 'low', 'normal' or 'high'"),
    ("complete_task", {"task_id": "3"}, "String should match pattern"),
    ("complete_task", {"task_id": "T-9"}, "No task with ID T-9. Call list_tasks"),
    ("complete_task", {}, "Field required"),
])
async def test_bad_calls_return_tool_errors(client, name, arguments, message):
    result = await client.call_tool(name, arguments)
    assert result.is_error
    assert message in result.content[0].text

The tests check three things: tools/list equals the snapshot exactly; each tool’s structuredContent validates against its own outputSchema and matches its text content; and each bad call returns a tool error with the expected message. The parametrized cases make every new argument rule one line to pin.

5. Add one integration test per transport

In-process tests cannot catch transport faults: a broken command line, stray output on stdout, a wrong endpoint path. One round trip per transport covers those. The http_url fixture starts serve.py --http on a free port, waits until the port accepts connections, and stops the process after the test.

"""Integration tests: the same server over stdio and over Streamable HTTP. Run: python -m pytest -v"""
import json
import socket
import subprocess
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path

import pytest
from mcp import Client, StdioServerParameters
from mcp.shared.exceptions import MCPError

import wire

pytestmark = pytest.mark.anyio
HERE = Path(__file__).resolve().parent
STDIO = StdioServerParameters(command=sys.executable, args=[str(HERE / "serve.py")])


@pytest.fixture
def anyio_backend():
    return "asyncio"


def free_port() -> int:
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]


@pytest.fixture
def http_url():
    """Start serve.py --http on a free port and return its MCP endpoint."""
    port = free_port()
    process = subprocess.Popen([sys.executable, "serve.py", "--http", "--port", str(port)], cwd=HERE, stderr=subprocess.DEVNULL)
    deadline = time.monotonic() + 15
    while socket.socket().connect_ex(("127.0.0.1", port)) != 0:
        assert time.monotonic() < deadline and process.poll() is None, "server did not start"
        time.sleep(0.1)
    yield f"http://127.0.0.1:{port}/mcp"
    process.terminate()
    process.wait(timeout=10)


async def round_trip(target) -> tuple:
    async with Client(target) as client:
        names = [tool.name for tool in (await client.list_tools()).tools]
        added = await client.call_tool("add_task", {"title": "Tag v1.2"})
        return names, added.structured_content["id"]


async def test_stdio_round_trip():
    assert await round_trip(STDIO) == (["list_tasks", "add_task", "complete_task"], "T-4")


async def test_streamable_http_round_trip(http_url):
    assert await round_trip(http_url) == (["list_tasks", "add_task", "complete_task"], "T-4")


async def test_http_clients_share_one_server_process(http_url):
    assert (await round_trip(http_url))[1] == "T-4"
    assert (await round_trip(http_url))[1] == "T-5"
    assert (await round_trip(STDIO))[1] == "T-4"  # each stdio client starts its own process


async def test_print_to_stdout_corrupts_the_stdio_stream(caplog):
    noisy = StdioServerParameters(command=sys.executable, cwd=str(HERE), args=[
        "-c", "print('tasks server starting'); import runpy; runpy.run_path('serve.py', run_name='__main__')"])
    assert (await round_trip(noisy))[0] == ["list_tasks", "add_task", "complete_task"]
    assert "Failed to parse JSONRPC message from server" in caplog.text


async def test_wrong_endpoint_path_is_not_found(http_url):
    with pytest.raises(ExceptionGroup) as raised:
        await round_trip(http_url.replace("/mcp", "/tasks"))
    assert raised.group_contains(MCPError, match="Not Found")


def test_foreign_host_header_is_rejected(http_url):
    request = urllib.request.Request(http_url, method="POST", data=b"{}", headers={
        "Host": "tasks.example.com", "Content-Type": "application/json", "Accept": "application/json, text/event-stream"})
    with pytest.raises(urllib.error.HTTPError) as raised:
        urllib.request.urlopen(request, timeout=10)
    assert (raised.value.code, raised.value.read()) == (421, b"Invalid Host header")


def test_client_script_prints_the_recorded_output(http_url):
    runs = [["stdio"], [http_url], [http_url]]
    printed = "".join(subprocess.run([sys.executable, "client.py", *arg], cwd=HERE, capture_output=True, text=True,
                                     timeout=60, check=True).stdout for arg in runs)
    assert printed == (HERE / "client-output.txt").read_text(encoding="utf-8")


def test_server_is_the_typed_tools_lesson_server():
    assert (HERE / "tasks_server.py").read_bytes() == (HERE.parent / "mcp-typed-tools" / "tasks_server.py").read_bytes()


def test_recorded_http_exchange_matches_a_fresh_run():
    assert wire.collect() == json.loads((HERE / "wire-exchange.json").read_text(encoding="utf-8"))

Keep integration tests few. Each one starts a process, so they are slower and they fail for reasons outside your code, such as a busy port. The schema and argument rules belong in the in-process suite.

.venv/bin/python -m pytest -v

Here is the finished test run: 19 tests in 5.93 seconds, including both transports and their failure cases.

============================= test session starts ==============================
collecting ... collected 19 items

test_contract.py::test_tools_list_matches_the_published_contract PASSED  [  5%]
test_contract.py::test_each_tool_returns_content_that_matches_its_output_schema[list_tasks-arguments0] PASSED [ 10%]
test_contract.py::test_each_tool_returns_content_that_matches_its_output_schema[list_tasks-arguments1] PASSED [ 15%]
test_contract.py::test_each_tool_returns_content_that_matches_its_output_schema[add_task-arguments2] PASSED [ 21%]
test_contract.py::test_each_tool_returns_content_that_matches_its_output_schema[complete_task-arguments3] PASSED [ 26%]
test_contract.py::test_bad_calls_return_tool_errors[add_task-arguments0-String should have at least 3 characters] PASSED [ 31%]
test_contract.py::test_bad_calls_return_tool_errors[add_task-arguments1-Input should be 'low', 'normal' or 'high'] PASSED [ 36%]
test_contract.py::test_bad_calls_return_tool_errors[complete_task-arguments2-String should match pattern] PASSED [ 42%]
test_contract.py::test_bad_calls_return_tool_errors[complete_task-arguments3-No task with ID T-9. Call list_tasks] PASSED [ 47%]
test_contract.py::test_bad_calls_return_tool_errors[complete_task-arguments4-Field required] PASSED [ 52%]
test_transports.py::test_stdio_round_trip PASSED                         [ 57%]
test_transports.py::test_streamable_http_round_trip PASSED               [ 63%]
test_transports.py::test_http_clients_share_one_server_process PASSED    [ 68%]
test_transports.py::test_print_to_stdout_corrupts_the_stdio_stream PASSED [ 73%]
test_transports.py::test_wrong_endpoint_path_is_not_found PASSED         [ 78%]
test_transports.py::test_foreign_host_header_is_rejected PASSED          [ 84%]
test_transports.py::test_client_script_prints_the_recorded_output PASSED [ 89%]
test_transports.py::test_server_is_the_typed_tools_lesson_server PASSED  [ 94%]
test_transports.py::test_recorded_http_exchange_matches_a_fresh_run PASSED [100%]

============================== 19 passed in 5.93s ==============================

The recorded test output appears above.

Inspect Streamable HTTP requests

wire.py records what the SDK client actually sends over Streamable HTTP. The SDK’s own transport makes every request; an httpx response hook, passed in through streamable_http_client(url, http_client=...), only copies each request and response into wire-exchange.json. Two call_tool calls produced 4 POSTs, all to /mcp: server/discover, tools/call, tools/list, tools/call.

The first request is server/discover, which asks for the server’s capabilities. There is no initialize handshake and no session: in protocol version 2026-07-28, every request carries its protocol version and client capabilities in params._meta, so the server can answer each POST on its own. The tools/list before the second call is the client checking results: call_tool validates structuredContent against the tool’s outputSchema, and it lists the tools once to cache those schemas. This is the failing call:

{
  "request": {
    "method": "POST",
    "path": "/mcp",
    "headers": {
      "accept": "application/json, text/event-stream",
      "content-type": "application/json",
      "mcp-protocol-version": "2026-07-28",
      "mcp-method": "tools/call",
      "mcp-name": "complete_task"
    },
    "body": {
      "jsonrpc": "2.0",
      "id": 4,
      "method": "tools/call",
      "params": {
        "name": "complete_task",
        "arguments": {
          "task_id": "T-9"
        },
        "_meta": {
          "io.modelcontextprotocol/protocolVersion": "2026-07-28",
          "io.modelcontextprotocol/clientInfo": {
            "name": "mcp",
            "version": "0.1.0"
          },
          "io.modelcontextprotocol/clientCapabilities": {}
        }
      }
    }
  },
  "response": {
    "status": 200,
    "contentType": "application/json",
    "body": {
      "jsonrpc": "2.0",
      "id": 4,
      "result": {
        "content": [
          {
            "text": "Error executing tool complete_task: No task with ID T-9. Call list_tasks to see the valid IDs.",
            "type": "text"
          }
        ],
        "isError": true,
        "resultType": "complete",
        "_meta": {
          "io.modelcontextprotocol/serverInfo": {
            "name": "tasks",
            "version": "1.0.0"
          }
        }
      }
    }
  }
}

Three headers repeat fields of the body: MCP-Protocol-Version, Mcp-Method and, for tools/call, Mcp-Name. They let a load balancer or gateway route and rate-limit by method or tool without parsing JSON. The specification requires the server to reject a request whose headers disagree with its body, with status 400 and error -32020 (HeaderMismatch), because a proxy that routes on the header while the server runs the body is a security hole. The reply here is a single JSON object with status 200 and Content-Type application/json; a server can instead stream progress notifications as SSE before the final result, and the specification requires clients to accept both. The tool error is an ordinary result with isError: true, exactly as in the in-process connection.

Over stdio the same JSON-RPC objects travel as single lines on the server’s stdin and stdout, without HTTP headers. That framing is why one stray print() breaks it.

Expose the HTTP transport safely

• Auth boundary. A stdio server inherits the access of the user who started it, and the protocol reaches it only through the stdin and stdout pipes its parent process opened; it listens on no port. That limits who can send requests, but it is not authentication: the server trusts whatever process launched it, and the operating system decides which other local processes of the same user can inspect or interfere with it. A Streamable HTTP server accepts anything that reaches its port. The SDK’s default allowlist rejects requests whose Host is not a localhost name, which blocks DNS-rebinding attacks from a browser, but it is not authentication. When you bind to another address, the SDK sets up no allowlist: pass transport_security and put an authorization layer in front of the tools. The remote MCP service lesson adds OAuth token checks.
• Shared state. The in-memory TASKS dict is shared by every HTTP client and lost on restart. A multi-client server keeps state in a store with its own concurrency control and scopes data to the caller. The specification’s guidance for state that must span calls is an explicit handle returned by one tool and passed to the next, because the protocol has no session.
• Timeouts. The SDK client waits without a limit by default: read_timeout_seconds is None unless you set it on Client or on a single call_tool. Set it in any client that must not hang on a slow tool.

Host header from another site. The SDK returns HTTP 421 with Invalid Host header before a tool runs. Keep that refusal for foreign hosts; behind a proxy, explicitly add the public host through transport_security.

Check your understanding

Build an MCP client with discovery, calls and tracing continues with the next part of the MCP application.

References

Official sources checked October 21, 2026, against mcp 2.3.0 and MCP specification 2026-07-28.

• MCP specification: transports and stdio
• MCP specification: Streamable HTTP, including request metadata headers, security and backward compatibility
• MCP specification: tools, including stateful tools
• MCP Python SDK repository and mcp 2.3.0 on PyPI
• pytest documentation and AnyIO: testing with AnyIO