12.3 Testing and Publishing

1. Unit Testing with Mocked APIs

Your search_pubchem tool passes every test on your laptop, you ship it on Friday afternoon, and by Monday morning an agent pipeline has silently returned empty results for 200 experiments because PubChem renamed a JSON field over the weekend. How do you build a test suite that would have caught that before the damage spread?

A server that ships without layered tests is one upstream API change away from silent data corruption, where an agent pipeline keeps running but every result it returns is wrong. The testing strategy in this section exists precisely to close that gap before it costs you weeks of contaminated experiments.

Mocking substitutes a real dependency (an HTTP client, a database connection, a file system call) with a controlled stand-in that returns predetermined responses. Scientific MCP tools typically depend on external APIs such as PubChem or OpenAlex. Calling those APIs in every test run introduces network latency, rate limit risk, and nondeterministic behavior when upstream data changes. Python's unittest.mock.patch intercepts the dependency's import path at runtime and replaces it with a Mock or AsyncMock object whose return values you define in the test. Use mocking for unit tests where you need speed and reproducibility; switch to live API calls (integration tests) when you need to verify that your code still matches the real service's current contract.

MCP tool testing operates at two levels. First, we test the tool function directly by calling it as a regular Python async function with mocked dependencies. Second, we test the JSON-RPC layer by constructing protocol messages and sending them through the server's message handler. The first level catches logic bugs; the second catches serialization and schema validation bugs. In short: if your tests never speak the same JSON-RPC protocol your agent speaks, you are testing a different system than the one running in production.

"""Unit tests for PubChem MCP tools."""
import json
import pytest
from unittest.mock import AsyncMock, patch

# Import the tool function directly
from science_mcp.tools.pubchem import search_pubchem, compute_descriptors


# --- Fixtures ---

@pytest.fixture
def mock_pubchem_search_response():
    """Pre-recorded PubChem search response for 'aspirin'."""
    return {
        "IdentifierList": {
            "CID": [2244, 71364, 71365]
        }
    }

@pytest.fixture
def mock_pubchem_properties_response():
    """Pre-recorded PubChem properties response for aspirin (CID 2244)."""
    return {
        "PropertyTable": {
            "Properties": [
                {
                    "CID": 2244,
                    "IUPACName": "2-acetyloxybenzoic acid",
                    "MolecularFormula": "C9H8O4",
                    "MolecularWeight": 180.16,
                    "CanonicalSMILES": "CC(=O)OC1=CC=CC=C1C(=O)O",
                }
            ]
        }
    }


# --- Tool logic tests ---

class TestSearchPubChem:
    """Tests for the search_pubchem tool."""

    @pytest.mark.asyncio
    async def test_successful_search(
        self, mock_pubchem_search_response, mock_pubchem_properties_response
    ):
        """A valid search returns structured compound data."""
        mock_responses = [
            AsyncMock(status_code=200, json=lambda: mock_pubchem_search_response),
            AsyncMock(status_code=200, json=lambda: mock_pubchem_properties_response),
        ]

        with patch("science_mcp.tools.pubchem.httpx.AsyncClient") as mock_client:
            mock_client.return_value.__aenter__ = AsyncMock(return_value=mock_client.return_value)
            mock_client.return_value.__aexit__ = AsyncMock(return_value=False)
            mock_client.return_value.get = AsyncMock(side_effect=mock_responses)

            result = await search_pubchem("aspirin", search_type="name", max_results=1)

        data = json.loads(result)
        assert "compounds" in data
        assert len(data["compounds"]) >= 1
        assert data["compounds"][0]["cid"] == 2244
        assert data["compounds"][0]["formula"] == "C9H8O4"
        assert data["source"] == "PubChem"

    @pytest.mark.asyncio
    async def test_compound_not_found(self):
        """A search for a nonexistent compound returns empty results."""
        mock_resp = AsyncMock(status_code=404)
        mock_resp.raise_for_status = AsyncMock(side_effect=None)

        with patch("science_mcp.tools.pubchem.httpx.AsyncClient") as mock_client:
            mock_client.return_value.__aenter__ = AsyncMock(return_value=mock_client.return_value)
            mock_client.return_value.__aexit__ = AsyncMock(return_value=False)
            mock_client.return_value.get = AsyncMock(return_value=mock_resp)

            result = await search_pubchem("xyznonexistent12345")

        data = json.loads(result)
        assert data["compounds"] == []
        assert data["total_count"] == 0

    @pytest.mark.asyncio
    async def test_invalid_max_results(self):
        """max_results outside allowed range raises ValueError."""
        with pytest.raises(ValueError, match="max_results must be 1-25"):
            await search_pubchem("aspirin", max_results=100)


class TestComputeDescriptors:
    """Tests for the compute_descriptors tool (no mocking needed)."""

    @pytest.mark.asyncio
    async def test_aspirin_descriptors(self):
        """Aspirin descriptors match known values."""
        result = await compute_descriptors("CC(=O)OC1=CC=CC=C1C(=O)O")
        data = json.loads(result)

        assert "descriptors" in data
        desc = data["descriptors"]
        assert 180 < desc["molecular_weight"] < 181  # ~180.16
        assert desc["hydrogen_bond_donors"] == 1
        assert desc["hydrogen_bond_acceptors"] == 4
        assert data["lipinski_rule_of_five"]["drug_like"] is True

    @pytest.mark.asyncio
    async def test_invalid_smiles(self):
        """Invalid SMILES returns a structured error, not an exception."""
        result = await compute_descriptors("not_a_molecule!!!")
        data = json.loads(result)

        assert "error" in data
        assert "Invalid SMILES" in data["error"]
        assert "suggestion" in data

2. Integration Testing Against Live APIs

Unit tests with mocks verify logic, but they cannot catch API changes (a renamed field, a new rate limit, a deprecated endpoint). Integration tests call the real API, but they are slower, non-deterministic, and require network access. Run integration tests in CI (Continuous Integration) on a schedule (nightly, not on every push) and mark them so developers can skip them locally.

"""Integration tests against live scientific APIs."""
import json
import pytest

from science_mcp.tools.pubchem import search_pubchem
from science_mcp.tools.literature import search_literature

# Mark all tests in this module as integration tests
pytestmark = pytest.mark.integration


class TestPubChemIntegration:
    """Live API tests for PubChem tools."""

    @pytest.mark.asyncio
    async def test_aspirin_search_live(self):
        """Search for aspirin returns the expected CID from live PubChem."""
        result = await search_pubchem("aspirin", search_type="name", max_results=3)
        data = json.loads(result)

        assert data["total_count"] >= 1
        cids = [c["cid"] for c in data["compounds"]]
        assert 2244 in cids, "Aspirin (CID 2244) should appear in results"

    @pytest.mark.asyncio
    async def test_smiles_search_live(self):
        """SMILES search returns matching compounds from live PubChem."""
        # Ethanol SMILES
        result = await search_pubchem("CCO", search_type="smiles", max_results=1)
        data = json.loads(result)

        assert data["total_count"] >= 1
        assert data["compounds"][0]["formula"] == "C2H6O"


class TestOpenAlexIntegration:
    """Live API tests for OpenAlex tools."""

    @pytest.mark.asyncio
    async def test_literature_search_live(self):
        """Search for a well-known topic returns results from live OpenAlex."""
        result = await search_literature(
            "transformer attention mechanism", max_results=5
        )
        data = json.loads(result)

        assert data["total_count"] > 0
        assert len(data["papers"]) <= 5
        # At least one paper should have a title
        assert any(p["title"] for p in data["papers"])

    @pytest.mark.asyncio
    async def test_date_filter_live(self):
        """Date filtering works correctly with live OpenAlex."""
        result = await search_literature(
            "CRISPR", max_results=3, year_from=2023, year_to=2024
        )
        data = json.loads(result)

        for paper in data["papers"]:
            if paper["year"]:
                assert 2023 <= paper["year"] <= 2024

Configure pytest to recognize the custom marker in pyproject.toml:

[tool.pytest.ini_options]
markers = [
    "integration: tests that call live external APIs (deselect with '-m not integration')",
]
asyncio_mode = "auto"

3. End-to-End Testing with the MCP Inspector

The MCP Inspector is a developer tool that connects to your server as a client and lets you interactively browse tools, call them, and inspect responses. It validates the JSON-RPC protocol, checks schema compliance, and displays the raw JSON-RPC message exchange. For automated end-to-end testing, we can script the inspector or use the MCP Python SDK's client to drive the server programmatically.

"""End-to-end tests using the MCP client SDK."""
import json
import pytest
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


@pytest.fixture
async def mcp_session():
    """Start the MCP server and connect a test client."""
    server_params = StdioServerParameters(
        command="python",
        args=["-m", "science_mcp.server"],
    )
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            yield session


class TestMCPProtocol:
    """End-to-end tests exercising the full JSON-RPC protocol."""

    @pytest.mark.asyncio
    async def test_list_tools(self, mcp_session):
        """Server advertises expected tools during capability negotiation."""
        tools = await mcp_session.list_tools()
        tool_names = {t.name for t in tools.tools}

        assert "search_pubchem" in tool_names
        assert "search_literature" in tool_names
        assert "compute_descriptors" in tool_names

    @pytest.mark.asyncio
    async def test_tool_schemas_valid(self, mcp_session):
        """Every advertised tool has a valid JSON Schema for its inputs."""
        tools = await mcp_session.list_tools()

        for tool in tools.tools:
            schema = tool.inputSchema
            assert schema.get("type") == "object", (
                f"Tool '{tool.name}' schema must have type 'object'"
            )
            assert "properties" in schema, (
                f"Tool '{tool.name}' schema must define properties"
            )

    @pytest.mark.asyncio
    async def test_call_compute_descriptors(self, mcp_session):
        """Calling compute_descriptors through the protocol returns valid JSON."""
        result = await mcp_session.call_tool(
            "compute_descriptors",
            arguments={"smiles": "CCO"},  # ethanol
        )

        # MCP returns content as a list of content blocks
        assert len(result.content) >= 1
        text = result.content[0].text
        data = json.loads(text)

        assert "descriptors" in data
        assert data["descriptors"]["molecular_weight"] > 0

    @pytest.mark.asyncio
    async def test_call_with_invalid_input(self, mcp_session):
        """Calling a tool with invalid input returns a structured error."""
        result = await mcp_session.call_tool(
            "compute_descriptors",
            arguments={"smiles": "NOT_VALID!!!"},
        )

        text = result.content[0].text
        data = json.loads(text)
        assert "error" in data

The three test layers form a pyramid (Figure 12.3): a broad base of fast unit tests, a narrower band of integration tests, and a small cap of end-to-end protocol tests. Each layer trades speed for realism, and skipping any one of them leaves an entire class of failure undetected.

Practical Example: The Three-Layer Test Pyramid

For a scientific MCP server with 10 tools, a healthy test distribution looks like this: 50+ unit tests (fast, mocked, run on every commit), 10-15 integration tests (live API, run nightly), and 5-10 end-to-end tests (full protocol, run on releases). The unit tests catch logic bugs in minutes. The integration tests catch API drift overnight. The end-to-end tests catch protocol-level regressions before users see them. This pyramid mirrors the testing strategies discussed in Chapter 18, and the CI/CD patterns from Chapter 21. Figure 12.3.1 illustrates MCP server three-layer test pyramid and CI pipeline.

4. Authentication and Authorization

Passing all three test layers confirms that your server behaves correctly, but correctness alone is not enough once the server is exposed beyond your local machine.

A scientific MCP server may expose tools that access proprietary databases, consume paid API quotas, or control expensive instruments. Authentication verifies who is connecting; authorization determines what they can do.

For stdio transport, authentication is implicit: the host spawns the server as a child process, and only the host can communicate with it. The server inherits the host's environment variables, which is the standard mechanism for passing API keys:

"""Authentication via environment variables (stdio transport)."""
import os
from mcp.server import Server

server = Server("authenticated-science")

def get_api_key(service: str) -> str:
    """Retrieve an API key from environment variables.

    Raises RuntimeError if the key is not set, with instructions
    for the user to configure it.
    """
    env_var = f"{service.upper()}_API_KEY"
    key = os.environ.get(env_var)
    if not key:
        raise RuntimeError(
            f"Missing {env_var} environment variable. "
            f"Set it in your MCP client configuration: "
            f'"env": {{"{env_var}": "your-key-here"}}'
        )
    return key

@server.tool()
async def search_proprietary_db(query: str) -> str:
    """Search a proprietary compound database (requires API key).

    Args:
        query: Search term for the database.
    """
    api_key = get_api_key("PROPRIETARY_DB")
    # Use the key to authenticate with the database
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            "https://api.proprietary-db.example.com/search",
            params={"q": query},
            headers={"Authorization": f"Bearer {api_key}"},
        )
        resp.raise_for_status()
        return resp.text

For HTTP transports (SSE (Server-Sent Events), Streamable HTTP), the server is a network service that anyone might attempt to reach. Here you need explicit authentication. The MCP specification recommends OAuth (Open Authorization) 2.0 for remote servers, with the host handling the OAuth flow and including the access token in HTTP headers.

The standard approach uses a bearer token (a credential string the client presents in an HTTP header to prove it has been authorized) validated by middleware that runs before the request reaches your tool handler.

"""Token-based auth middleware for HTTP-transport MCP servers."""
from functools import wraps
from starlette.requests import Request
from starlette.responses import JSONResponse

def require_auth(allowed_scopes: list[str]):
    """Middleware that validates OAuth bearer tokens and scopes.

    Args:
        allowed_scopes: List of OAuth scopes required to access this endpoint.
    """
    def decorator(func):
        @wraps(func)
        async def wrapper(request: Request, *args, **kwargs):
            auth_header = request.headers.get("Authorization", "")
            if not auth_header.startswith("Bearer "):
                return JSONResponse(
                    {"error": "Missing or invalid Authorization header"},
                    status_code=401,
                )

            token = auth_header[7:]
            # Validate the token (in production, verify the JWT (JSON Web Token) signature and claims)
            # validate_token is an application-specific function you implement
            # to decode the JWT, check its signature, and return the claims dict
            claims = await validate_token(token)
            if claims is None:
                return JSONResponse(
                    {"error": "Invalid or expired token"},
                    status_code=401,
                )

            # Check scopes
            token_scopes = set(claims.get("scope", "").split())
            if not token_scopes.issuperset(allowed_scopes):
                return JSONResponse(
                    {"error": f"Insufficient scope. Required: {allowed_scopes}"},
                    status_code=403,
                )

            return await func(request, *args, **kwargs)
        return wrapper
    return decorator

5. Sandboxing with Docker

Authentication controls who can reach your server, but it does not limit what a legitimately authenticated request can do to the host system.

MCP servers that execute user-provided input (SMILES strings, search queries, code snippets) should run in a sandbox. Docker enforces process isolation, filesystem and network restrictions, and resource limits, preventing the container from reaching the host filesystem, making arbitrary connections, or consuming unbounded CPU and memory.

# Dockerfile for a sandboxed scientific MCP server
FROM python:3.12-slim

# Install system dependencies for RDKit
RUN apt-get update && apt-get install -y --no-install-recommends \
    libxrender1 libxext6 \
    && rm -rf /var/lib/apt/lists/*

# Create non-root user
RUN useradd --create-home --shell /bin/bash mcpuser
WORKDIR /home/mcpuser/app

# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy server code
COPY src/ ./src/

# Switch to non-root user
USER mcpuser

# Run server on stdio (default) or HTTP
ENV MCP_TRANSPORT=stdio
ENTRYPOINT ["python", "-m", "science_mcp.server"]

# docker-compose.yml for MCP server with resource limits
services:
  science-mcp:
    build: .
    environment:
      - PUBCHEM_API_KEY=${PUBCHEM_API_KEY}
      - OPENALEX_EMAIL=${OPENALEX_EMAIL}
    deploy:
      resources:
        limits:
          cpus: "2.0"
          memory: 1G
        reservations:
          cpus: "0.5"
          memory: 256M
    # Restrict network access to only scientific APIs
    networks:
      - science-apis
    # Read-only filesystem except for tmp
    read_only: true
    tmpfs:
      - /tmp:size=100M

networks:
  science-apis:
    driver: bridge

6. Publishing and Distribution

A well-tested, sandboxed MCP server is ready for distribution. The MCP ecosystem supports several distribution channels:

PyPI (Python Package Index) is the standard channel for Python packages. Package your server as a regular Python package with pyproject.toml, and users install it with pip install science-mcp. The entry point should be a console script that starts the server on stdio. To publish, build the distribution with python -m build and upload it with twine upload dist/*; both tools are installable via pip install build twine.

# pyproject.toml for an MCP server package
[project]
name = "science-mcp"
version = "1.0.0"
description = "MCP server for scientific discovery workflows"
requires-python = ">=3.11"
dependencies = [
    "mcp>=1.0.0",
    "httpx>=0.27.0",
    "pydantic>=2.0",
    "pubchempy>=1.0.4",
]

[project.optional-dependencies]
chemistry = ["rdkit-pypi>=2024.3.1"]
all = ["science-mcp[chemistry]"]

[project.scripts]
science-mcp = "science_mcp.server:main_sync"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

Claude Desktop configuration is where most users will connect to your server. The configuration file maps server names to launch commands:

{
  "mcpServers": {
    "science-mcp": {
      "command": "science-mcp",
      "env": {
        "OPENALEX_EMAIL": "researcher@university.edu"
      }
    },
    "science-mcp-docker": {
      "command": "docker",
      "args": ["run", "-i", "--rm", "science-mcp:latest"],
      "env": {
        "OPENALEX_EMAIL": "researcher@university.edu"
      }
    }
  }
}

7. Continuous Integration Pipeline

With the server packaged and listed in a registry, the final piece is automating the test and release cycle so that every change is validated before it reaches users.

A complete CI pipeline for an MCP server runs unit tests on every push, integration tests nightly, and end-to-end protocol tests on every release. Here is a GitHub Actions workflow that implements this:

# .github/workflows/test.yml
name: Test MCP Server

on:
  push:
    branches: [main]
  pull_request:
  schedule:
    - cron: "0 2 * * *"  # Nightly at 2 AM UTC

jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install -e ".[all]" && pip install pytest pytest-asyncio
      - run: pytest -m "not integration" --tb=short

  integration-tests:
    if: github.event_name == 'schedule' || github.event_name == 'push'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install -e ".[all]" && pip install pytest pytest-asyncio
      - run: pytest -m integration --tb=short
        env:
          OPENALEX_EMAIL: ci@example.com

  protocol-tests:
    if: startsWith(github.ref, 'refs/tags/v')
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install -e ".[all]" && pip install pytest pytest-asyncio mcp
      - run: pytest tests/test_protocol.py --tb=short

Exercises

1. Conceptual: Explain why end-to-end MCP protocol tests are necessary in addition to unit tests that call tool functions directly. What class of bugs can protocol tests catch that unit tests cannot? Give two concrete examples.
2. Coding: Write a pytest fixture that records HTTP responses from PubChem to JSON files on the first run, then replays them on subsequent runs (a "cassette" pattern, named by analogy with recording and replaying audio cassettes). Use the pytest-recording or vcrpy library, or implement it from scratch with httpx event hooks. Verify that your search_pubchem tests produce identical results in both live and replay modes.
3. Analysis: The Docker configuration in this section limits the server to 1 GB of memory. Estimate the maximum number of concurrent compute_descriptors calls the server can handle before hitting this limit, assuming each RDKit molecule object consumes approximately 5 KB of memory and each call holds the molecule for 200 ms. What happens when the limit is exceeded, and how should the server handle it?