Integration tests for MCP flows

The quick answer

An integration test drives your MCP server the way a client actually does: it opens a session, runs the initialize handshake, lists what the server offers, then calls tools in sequence and checks that each step feeds the next. (That's the flow for every spec revision through 2025-11-25, which is what the pinned SDK below speaks. The 2026-07-28 revision removes the initialize handshake entirely — there's no negotiation step at all, just per-request _meta and an optional server/discover call — so a client written against that revision skips straight to listing and calling.) The trick is to skip sockets entirely. The official SDKs ship an in-memory transport that wires a real client to your real server object in the same process, so you get the whole protocol path without a port, a subprocess, or a flaky network.

In Python, the helper is create_connected_server_and_client_session. It takes your FastMCP instance, runs the handshake for you, and hands back a live ClientSession.

import pytest
from mcp.server.fastmcp import FastMCP
from mcp.shared.memory import create_connected_server_and_client_session as connect

@pytest.mark.asyncio
async def test_list_then_call():
    server = FastMCP("demo")

    @server.tool()
    def greet(name: str) -> str:
        return f"hi {name}"

    async with connect(server) as client:
        tools = await client.list_tools()
        assert "greet" in {t.name for t in tools.tools}

        result = await client.call_tool("greet", {"name": "Ada"})
        assert result.content[0].text == "hi Ada"

That session already ran initialize before it yielded, so by the time you call list_tools you're exercising the real handshake, the real registry, and the real serialization path. The rest of this guide builds that into a multi-step flow, shows the same pattern in TypeScript, and covers the one assertion that trips people up: a tool that fails does not raise on the client.

Where this differs from unit testing

A unit test pins one handler in place and checks it in isolation. That's the right tool when you want to know whether greet formats its string correctly. If that's what you're after, start with writing unit tests for MCP servers, which stays inside a single tool.

Integration testing asks a different question: does a sequence of calls hold together? Real MCP usage is rarely one call. A model lists tools, calls one to get an id, passes that id into the next call, then reads a resource that reflects the result. Each step depends on state the previous step created. That chain is exactly what breaks in ways a single-handler test can't catch: an id that serializes wrong, state that doesn't persist between calls, an error in step two that leaves step three reading stale data. You test the flow by running the flow.

Prerequisites

• Python 3.10+ (this guide uses the official mcp package) or Node.js 18+ (@modelcontextprotocol/sdk)
• A test runner: pip install pytest pytest-asyncio for Python, or npm install -D vitest for TypeScript
• A server you can import into the test process. In-memory testing binds to the server object directly, so the server has to be constructable in code, not just launchable as a script.

Install the Python pieces:

$pip install "mcp>=1.27,<2" pytest pytest-asyncio

The <2 pin keeps you on the stable v1 line, where the in-memory helper lives at mcp.shared.memory and the server class is FastMCP (python-sdk testing docs). mcp 2.0.0 shipped stable on 2026-07-28, the same day as the 2026-07-28 spec revision, and is what an unpinned pip install mcp gets today — it's not a pre-release you can safely ignore. It renames FastMCP to MCPServer (mcp.server.fastmcp → mcp.server.mcpserver) and removes create_connected_server_and_client_session outright in favor of a unified mcp.client.Client, so the <2 pin above is what keeps this guide's examples working as written. See the migration guide if you're porting them to 2.0.0.

A multi-step flow in Python

Here's a small cart server with three tools that depend on each other: create a cart, add items to it, then check out. Returning typed values (a Pydantic model) is what makes the SDK populate structuredContent on the result, so the test can read fields by name instead of parsing text.

from pydantic import BaseModel
from mcp.server.fastmcp import FastMCP

class Cart(BaseModel):
    id: str
    items: list[str]
    total: float

class Order(BaseModel):
    order_id: str
    total: float
    status: str

def build_server() -> FastMCP:
    server = FastMCP("cart-server")
    carts: dict[str, Cart] = {}

    @server.tool()
    def create_cart() -> Cart:
        cart_id = f"cart-{len(carts) + 1}"
        carts[cart_id] = Cart(id=cart_id, items=[], total=0.0)
        return carts[cart_id]

    @server.tool()
    def add_item(cart_id: str, sku: str, price: float) -> Cart:
        cart = carts[cart_id]
        cart.items.append(sku)
        cart.total = round(cart.total + price, 2)
        return cart

    @server.tool()
    def checkout(cart_id: str) -> Order:
        cart = carts[cart_id]
        if not cart.items:
            raise ValueError("cannot checkout an empty cart")
        return Order(order_id=f"order-{cart_id}", total=cart.total, status="confirmed")

    return server

The test connects once and walks the whole flow, threading each step's output into the next:

import pytest
from mcp.shared.memory import create_connected_server_and_client_session as connect

@pytest.mark.asyncio
async def test_checkout_flow():
    async with connect(build_server()) as client:
        listed = await client.list_tools()
        assert {t.name for t in listed.tools} == {"create_cart", "add_item", "checkout"}

        created = await client.call_tool("create_cart", {})
        cart_id = created.structuredContent["id"]

        await client.call_tool("add_item", {"cart_id": cart_id, "sku": "A1", "price": 9.99})
        updated = await client.call_tool("add_item", {"cart_id": cart_id, "sku": "B2", "price": 5.01})
        assert updated.structuredContent["items"] == ["A1", "B2"]
        assert updated.structuredContent["total"] == 15.0

        order = await client.call_tool("checkout", {"cart_id": cart_id})
        assert order.structuredContent["status"] == "confirmed"
        assert order.structuredContent["total"] == 15.0

This catches what a unit test wouldn't: the cart_id has to survive a round trip through the protocol and come back usable, the carts dict has to hold state across three separate calls on one session, and the running total is only right if both add_item calls really did land in order against shared state. If any of that breaks, this test fails where an isolated handler test would still pass.

One thing to watch: if a tool isn't annotated with a typed return, structuredContent comes back empty and you read result.content[0].text instead.

The assertion that trips people up

When a tool raises, the client does not see an exception. The SDK catches it inside the server and returns a normal result with isError set to true and the message tucked into the text content. So the natural-looking with pytest.raises(...) around a failing call_tool will itself fail, because nothing is raised.

Assert on the result instead:

@pytest.mark.asyncio
async def test_empty_cart_is_reported_as_tool_error():
    async with connect(build_server()) as client:
        created = await client.call_tool("create_cart", {})
        cart_id = created.structuredContent["id"]

        result = await client.call_tool("checkout", {"cart_id": cart_id})
        assert result.isError is True
        assert "empty cart" in result.content[0].text

This is the same two-channel behavior the spec describes: an error inside your tool logic comes back as a result with isError: true so the model can read it and react, while protocol-level problems like an unknown tool travel as JSON-RPC errors (MCP tools spec). The error handling in custom MCP servers guide covers that split in full. For a flow test, the rule of thumb is simple: a tool that fails is a result to inspect, not an exception to catch.

You might reach for connect(server, raise_exceptions=True) here, but it won't change what you see. That flag only affects lower-level protocol and handler errors; an exception thrown inside a tool body is always caught and converted to an isError result before the flag is ever consulted. So a failing tool still comes back as a result to inspect, not something you can catch, and your assertions stay exactly the same whether the flag is set or not.

The same flow in TypeScript

The TypeScript SDK ships InMemoryTransport.createLinkedPair(), which gives you two ends of an in-memory channel. Connect your server to one and a Client to the other, and you have the same in-process session. Register tools with registerTool, and pass an outputSchema when you want structuredContent back.

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js";
import { z } from "zod";
import { expect, test } from "vitest";

test("checkout flow", async () => {
  const server = new McpServer({ name: "cart-server", version: "1.0.0" });
  const carts = new Map<string, { id: string; items: string[]; total: number }>();

  server.registerTool(
    "create_cart",
    { inputSchema: {}, outputSchema: { id: z.string(), items: z.array(z.string()), total: z.number() } },
    async () => {
      const id = `cart-${carts.size + 1}`;
      const cart = { id, items: [] as string[], total: 0 };
      carts.set(id, cart);
      return { content: [], structuredContent: cart };
    }
  );

  server.registerTool(
    "checkout",
    { inputSchema: { cartId: z.string() } },
    async ({ cartId }) => {
      const cart = carts.get(cartId)!;
      if (cart.items.length === 0) {
        return { content: [{ type: "text", text: "cannot checkout an empty cart" }], isError: true };
      }
      return { content: [{ type: "text", text: `order-${cartId} confirmed` }] };
    }
  );

  const [clientEnd, serverEnd] = InMemoryTransport.createLinkedPair();
  const client = new Client({ name: "test", version: "1.0.0" });
  await Promise.all([server.connect(serverEnd), client.connect(clientEnd)]);

  const created = await client.callTool({ name: "create_cart", arguments: {} });
  const cartId = (created.structuredContent as { id: string }).id;

  const failed = await client.callTool({ name: "checkout", arguments: { cartId } });
  expect(failed.isError).toBe(true);
});

The registerTool call is the current API. The older server.tool() form has already been removed, not just deprecated — the new, split @modelcontextprotocol/server package hit 2.0.0 on 2026-07-27, a day ahead of the 2026-07-28 spec revision, and dropped .tool(), .prompt(), and .resource() for good, in favor of registerTool/registerPrompt/registerResource. This guide's examples use @modelcontextprotocol/sdk, the pre-split v1 monolith (still npm's latest tag), which still ships the deprecated .tool() form alongside registerTool for now. (TypeScript SDK) The error behavior matches Python exactly: a tool that returns isError: true shows up as result.isError on the client, not as a thrown error, so you assert on the result the same way.

Poking a flow by hand with the Inspector

Before you write a test, it helps to run the flow interactively and see the raw responses. The MCP Inspector has a CLI mode that calls a single method and prints the result, which is handy for confirming a tool's exact output shape before you assert on it.

List what a server exposes:

$npx @modelcontextprotocol/inspector --cli python server.py --method tools/list

Call one tool with arguments:

$npx @modelcontextprotocol/inspector --cli python server.py --method tools/call --tool-name create_cart

Each --tool-arg key=value adds one argument, and you repeat the flag for more (Inspector). This is a fast way to eyeball a response during development, but it runs one call at a time and spins up a fresh session each invocation, so it can't test a chain where step two depends on step one's output. That's what the in-memory session in your test suite is for: one connection, many calls, shared state.

Testing resources in a flow

Tools aren't the only surface a flow touches. Many servers write through a tool and read back through a resource, and that read path deserves a test too. On the same in-memory session, read_resource takes a URI and returns content you assert on.

@pytest.mark.asyncio
async def test_write_then_read_resource():
    server = FastMCP("notes-server")
    notes: dict[str, str] = {}

    @server.tool()
    def save_note(note_id: str, body: str) -> str:
        notes[note_id] = body
        return f"saved {note_id}"

    @server.resource("note://{note_id}")
    def read_note(note_id: str) -> str:
        return notes[note_id]

    async with connect(server) as client:
        await client.call_tool("save_note", {"note_id": "n1", "body": "hello world"})
        result = await client.read_resource("note://n1")
        assert result.contents[0].text == "hello world"

This is a genuine two-surface flow: the tool call mutates state, and the resource read reflects it through a different code path. The .contents list holds the returned blocks, and .text is where a text resource's body lands.

Transport notes

In-memory testing deliberately bypasses the transport layer, which is the point: your flow tests shouldn't depend on ports or timing. That means these tests validate your handler logic, tool wiring, and state handling, but not your HTTP setup.

If your server runs over Streamable HTTP in production, the single-endpoint transport that's been the standard since the 2025-03-26 spec revision, keep a thin layer of transport-level checks separate from these flow tests: that the MCP-Protocol-Version header round-trips, that auth rejects an unauthenticated call, and — for a server on the 2026-07-28 revision — that a POST to /mcp gets back the right Content-Type (application/json or text/event-stream) and that server/discover responds. (Older guidance said to check that a request minted a session id via Mcp-Session-Id. The 2026-07-28 revision removes protocol-level sessions entirely, so a modern server mints no session id to check for; only a server still speaking 2025-11-25 or earlier will set that header.) Those belong in their own suite, close to your deployment. If you see references to a two-endpoint HTTP plus SSE setup, that's the older transport that's been deprecated since 2025-03-26 and shouldn't shape new tests (transports spec). For the flow logic itself, the in-memory session is both faster and more honest, because it removes the network as a source of false failures.

Where AgentCat fits

Tests tell you a flow works against inputs you thought to write. Production tells you which flows real clients actually run, in what order, and where they stall. AgentCat records those real tool-call sequences from your live server, so the multi-step paths you saw agents take become the flows you turn into integration tests. The gap between "the tests I wrote" and "the flows users run" is usually where the interesting bugs live.

Next steps

• Writing unit tests for MCP servers for pinning individual handlers in isolation
• Validation tests for tool inputs for the argument-checking layer these flow tests assume is in place
• Error handling in custom MCP servers for the full isError versus JSON-RPC error split