12.2 Implementing Scientific Tools

1. Tool Design Principles

What happens when an agent sends a malformed SMILES string to PubChem, parses the HTML error page as valid JSON, and confidently reports fictional molecular properties to the researcher who asked for help? Four design principles prevent that scenario, and they apply regardless of whether you are wrapping a chemistry database, a genomics API, or a telescope control system.

Principle 1: Narrow scope, rich description. Each tool should do one thing well. A tool called search_compounds should search for compounds, not also compute properties, visualize structures, and predict toxicity. Narrow scope helps the LLM select the right tool. Descriptions, by contrast, should be generous: include input formats, example values, error conditions, and output structure. The LLM is the sole reader of these descriptions.

Principle 2: Validate early, fail clearly. Every input should be validated before any external API call is made. If a SMILES string is syntactically invalid, report that immediately rather than sending it to PubChem and parsing a cryptic HTTP 400 response. Error messages should be actionable: "Invalid SMILES: unmatched parenthesis at position 7" is useful; "Bad request" is not.

Principle 3: Return structured data. Tools should return JSON strings that the LLM can parse, not free-text descriptions. Include units, identifiers, and provenance information. When the agent chains multiple tools (search a compound, then compute its properties, then find related literature), each step must derive its input from the previous step's output. Structured data makes this chaining reliable. This principle connects to the knowledge representation ideas from Chapter 3: the tool's output is a knowledge artifact that must be machine-readable.

Principle 4: Respect rate limits. Scientific APIs are shared resources. PubChem allows roughly 5 requests per second; OpenAlex allows around 10 (check each API's current documentation, as these limits change). (A database of 116 million compounds, accessible through a pipe no wider than five queries per second: without client-side throttling, a single eager agent can lock itself out in under a second.) A well-behaved MCP server enforces these limits internally rather than relying on the host or the agent to throttle. Rate limiting also protects your server from an overeager agent that decides to enumerate every compound in a database.

How the Rate Limiter Works

A rate limiter controls how many requests your code sends to an external service within a given time window. It prevents your application from being blocked or banned for excessive usage. Scientific APIs like PubChem serve millions of researchers worldwide. Exceeding their posted limits triggers HTTP 429 errors that stall your entire agent pipeline, and repeated violations can lead to IP bans. The token-bucket algorithm maintains a counter of available "tokens" that refills at a steady rate (e.g., 5 per second). Each request consumes one token; if none are available, the caller sleeps until one regenerates. Use a rate limiter whenever your MCP server wraps an external API. For purely local computation tools (like the RDKit descriptor calculator below), you control the resource entirely, so no rate limiting is needed.

These principles become concrete when applied to real scientific APIs, so let us start with the most common pattern: querying a public database for molecular data.

2. Pattern 1: Database Lookup (PubChem)

When a research team's automated screening pipeline queries the wrong compound or silently drops results because of a malformed API call, weeks of downstream wet-lab work can be wasted chasing false leads. Getting the implementation right at the tool boundary is what separates an agent that accelerates discovery from one that poisons it.

PubChem is widely regarded as the largest open chemistry database, containing over 116 million compounds (circa 2024) with associated bioactivity data. Its Power User Gateway (PUG) REST API provides programmatic access to compound searches, property lookups, and similarity queries. Three endpoints become MCP tools here: compound search by name, property retrieval by CID (compound identifier), and similarity search by structure. In short: a scientific tool is a contract between the agent and reality; validate the input, structure the output, and the agent becomes a reliable lab partner.

"""PubChem MCP tools with rate limiting and structured output."""
import asyncio
import json
import time
from typing import Literal

import httpx
from mcp.server import Server
from pydantic import BaseModel, Field

PUBCHEM_BASE = "https://pubchem.ncbi.nlm.nih.gov/rest/pug"

# Rate limiter: PubChem allows 5 requests/second
class RateLimiter:
    """Token-bucket rate limiter for API calls."""

    def __init__(self, rate: float, burst: int = 1):
        self.rate = rate          # tokens per second
        self.burst = burst        # max tokens
        self.tokens = float(burst)
        self.last_refill = time.monotonic()
        self._lock = asyncio.Lock()

    async def acquire(self):
        async with self._lock:
            now = time.monotonic()
            elapsed = now - self.last_refill
            self.tokens = min(self.burst, self.tokens + elapsed * self.rate)
            self.last_refill = now
            if self.tokens < 1.0:
                wait = (1.0 - self.tokens) / self.rate
                await asyncio.sleep(wait)
                self.tokens = 0.0
            else:
                self.tokens -= 1.0

pubchem_limiter = RateLimiter(rate=5.0, burst=5)

server = Server("pubchem-tools")

@server.tool()
async def search_pubchem(
    query: str,
    search_type: Literal["name", "formula", "smiles"] = "name",
    max_results: int = Field(default=5, ge=1, le=25),
) -> str:
    """Search PubChem for chemical compounds.

    Finds compounds matching a name, molecular formula, or SMILES string
    and returns their identifiers and basic properties.

    Args:
        query: The search term. Examples by search_type:
               - name: 'aspirin', 'caffeine', 'glucose'
               - formula: 'C9H8O4', 'C8H10N4O2'
               - smiles: 'CC(=O)OC1=CC=CC=C1C(=O)O'
        search_type: How to interpret the query string.
        max_results: Number of results to return (1-25).

    Returns:
        JSON object with 'compounds' array and 'total_count' field.
        Each compound has: cid, name, formula, weight, smiles.
    """
    await pubchem_limiter.acquire()

    namespace_map = {"name": "name", "formula": "fastformula", "smiles": "smiles"}
    namespace = namespace_map[search_type]

    async with httpx.AsyncClient(timeout=30.0) as client:
        # Step 1: Search for CIDs
        search_url = f"{PUBCHEM_BASE}/compound/{namespace}/{query}/cids/JSON"
        resp = await client.get(search_url)

        if resp.status_code == 404:
            return json.dumps({"compounds": [], "total_count": 0,
                               "message": f"No compounds found for '{query}'"})
        resp.raise_for_status()

        cids = resp.json()["IdentifierList"]["CID"][:max_results]

        # Step 2: Fetch properties for matched CIDs
        await pubchem_limiter.acquire()
        cid_str = ",".join(str(c) for c in cids)
        props_url = (
            f"{PUBCHEM_BASE}/compound/cid/{cid_str}/property/"
            "IUPACName,MolecularFormula,MolecularWeight,CanonicalSMILES/JSON"
        )
        props_resp = await client.get(props_url)
        props_resp.raise_for_status()

        properties = props_resp.json()["PropertyTable"]["Properties"]
        compounds = [
            {
                "cid": p["CID"],
                "name": p.get("IUPACName", "unknown"),
                "formula": p.get("MolecularFormula", ""),
                "weight": p.get("MolecularWeight", 0.0),
                "smiles": p.get("CanonicalSMILES", ""),
            }
            for p in properties
        ]

    return json.dumps({
        "compounds": compounds,
        "total_count": len(cids),
        "source": "PubChem",
        "query": query,
        "search_type": search_type,
    }, indent=2)

The code above reflects several deliberate design choices. First, the rate limiter is shared across all tool calls in the server process, not per-tool. If the server exposes three PubChem tools, they all draw from the same 5-requests-per-second budget. Second, the tool makes two sequential API calls (search, then property fetch), each rate-limited independently. Third, the output includes provenance fields (source, query, search_type) that help the agent (and the human reviewing the agent's work) trace where the data came from. Provenance tracking becomes critical in Chapter 47: Experiment Registries.

3. Pattern 2: Literature Search (OpenAlex)

OpenAlex is an open catalog of scholarly works, authors, institutions, and concepts covering over 250 million publications (circa 2024). Its REST API supports full-text search, filtering by date, author, journal, and concept, and returns structured metadata including abstracts, citation counts, and open-access URLs.

"""OpenAlex MCP tools for literature search."""
import json
from typing import Literal

import httpx
from mcp.server import Server
from pydantic import Field

server = Server("literature-tools")

OPENALEX_BASE = "https://api.openalex.org"
# OpenAlex polite pool: include your email for higher rate limits
OPENALEX_EMAIL = "your-email@institution.edu"

@server.tool()
async def search_literature(
    query: str,
    max_results: int = 10,
    sort_by: Literal["relevance", "cited_by_count", "publication_date"] = "relevance",
    year_from: int | None = None,
    year_to: int | None = None,
    open_access_only: bool = False,
) -> str:
    """Search scholarly literature using OpenAlex.

    Finds academic papers matching a text query, with optional filters
    for date range, citation count, and open access availability.

    Args:
        query: Search terms (e.g., 'CRISPR gene editing efficiency').
               Supports boolean operators: 'machine learning AND drug discovery'.
        max_results: Number of papers to return (1-50, default 10).
        sort_by: Sort order for results.
        year_from: Filter to papers published on or after this year (e.g., 2020).
        year_to: Filter to papers published on or before this year (e.g., 2026).
        open_access_only: If true, return only open-access papers.

    Returns:
        JSON with 'papers' array (title, authors, year, doi, abstract, citations,
        open_access_url) and 'total_count' of matching works.
    """
    params = {
        "search": query,
        "per_page": min(max_results, 50),
        "mailto": OPENALEX_EMAIL,
    }

    # Build sort parameter
    sort_map = {
        "relevance": "relevance_score:desc",
        "cited_by_count": "cited_by_count:desc",
        "publication_date": "publication_date:desc",
    }
    params["sort"] = sort_map[sort_by]

    # Build filter string
    filters = []
    if year_from:
        filters.append(f"from_publication_date:{year_from}-01-01")
    if year_to:
        filters.append(f"to_publication_date:{year_to}-12-31")
    if open_access_only:
        filters.append("is_oa:true")
    if filters:
        params["filter"] = ",".join(filters)

    async with httpx.AsyncClient(timeout=30.0) as client:
        resp = await client.get(f"{OPENALEX_BASE}/works", params=params)
        resp.raise_for_status()
        data = resp.json()

    papers = []
    for work in data.get("results", []):
        # Extract first 3 author names
        authors = [
            a.get("author", {}).get("display_name", "Unknown")
            for a in work.get("authorships", [])[:3]
        ]
        if len(work.get("authorships", [])) > 3:
            authors.append("et al.")

        # Extract abstract from inverted index
        abstract = _reconstruct_abstract(work.get("abstract_inverted_index"))

        papers.append({
            "title": work.get("title", "Untitled"),
            "authors": authors,
            "year": work.get("publication_year"),
            "doi": work.get("doi"),
            "abstract": abstract[:500] if abstract else None,
            "cited_by_count": work.get("cited_by_count", 0),
            "open_access_url": work.get("open_access", {}).get("oa_url"),
            "openalex_id": work.get("id"),
        })

    return json.dumps({
        "papers": papers,
        "total_count": data.get("meta", {}).get("count", 0),
        "source": "OpenAlex",
        "query": query,
    }, indent=2)


def _reconstruct_abstract(inverted_index: dict | None) -> str | None:
    """Reconstruct abstract text from OpenAlex inverted index format.

    An inverted index maps each word to a list of positions where it appears,
    so {"the": [0, 5], "cell": [1]} means "the" is at positions 0 and 5 and
    "cell" is at position 1. Reversing this mapping recovers the original text.
    """
    if not inverted_index:
        return None
    # OpenAlex stores abstracts as {word: [positions]}
    word_positions = []
    for word, positions in inverted_index.items():
        for pos in positions:
            word_positions.append((pos, word))
    word_positions.sort()
    return " ".join(word for _, word in word_positions)

4. Pattern 3: Local Computation (RDKit)

Not every tool calls an external API. Many scientific workflows require local computation: calculating molecular descriptors, parsing protein sequences, running statistical tests, or transforming coordinate systems. These tools are faster (no network latency), more reliable (no API downtime), and can process sensitive data that should not leave the local machine.

"""Local chemistry computation tools using RDKit."""
import json
from mcp.server import Server

server = Server("chemistry-compute")

@server.tool()
async def compute_descriptors(smiles: str) -> str:
    """Compute physicochemical descriptors for a molecule.

    Calculates drug-likeness properties commonly used in medicinal chemistry:
    molecular weight, LogP (lipophilicity), number of hydrogen bond donors
    and acceptors, topological polar surface area (TPSA), and number of
    rotatable bonds.

    Args:
        smiles: Molecule in SMILES notation.
                Examples: 'CCO' (ethanol), 'CC(=O)OC1=CC=CC=C1C(=O)O' (aspirin).

    Returns:
        JSON with molecular descriptors and Lipinski Rule of Five assessment.
        Lipinski violations > 1 suggest poor oral bioavailability.
    """
    from rdkit import Chem
    from rdkit.Chem import Descriptors, Crippen, Lipinski, rdMolDescriptors

    mol = Chem.MolFromSmiles(smiles)
    if mol is None:
        return json.dumps({
            "error": f"Invalid SMILES: '{smiles}'. Could not parse molecule.",
            "suggestion": "Check for unmatched parentheses, invalid atom symbols, "
                          "or incorrect bond notation.",
        })

    # Compute descriptors
    mw = Descriptors.MolWt(mol)
    logp = Crippen.MolLogP(mol)
    hbd = Lipinski.NumHDonors(mol)
    hba = Lipinski.NumHAcceptors(mol)
    tpsa = rdMolDescriptors.CalcTPSA(mol)
    rotatable = Lipinski.NumRotatableBonds(mol)
    rings = Lipinski.RingCount(mol)
    heavy_atoms = mol.GetNumHeavyAtoms()

    # Lipinski Rule of Five assessment
    violations = sum([
        mw > 500,
        logp > 5,
        hbd > 5,
        hba > 10,
    ])

    return json.dumps({
        "smiles": smiles,
        "canonical_smiles": Chem.MolToSmiles(mol),
        "descriptors": {
            "molecular_weight": round(mw, 2),
            "logP": round(logp, 2),
            "hydrogen_bond_donors": hbd,
            "hydrogen_bond_acceptors": hba,
            "topological_polar_surface_area": round(tpsa, 2),
            "rotatable_bonds": rotatable,
            "ring_count": rings,
            "heavy_atom_count": heavy_atoms,
        },
        "lipinski_rule_of_five": {
            "violations": violations,
            "drug_like": violations <= 1,
            "details": {
                "MW_le_500": mw <= 500,
                "LogP_le_5": logp <= 5,
                "HBD_le_5": hbd <= 5,
                "HBA_le_10": hba <= 10,
            },
        },
    }, indent=2)


@server.tool()
async def find_similar_compounds(
    smiles: str,
    threshold: float = 0.7,
    fingerprint: str = "morgan",
) -> str:
    """Find structurally similar compounds using molecular fingerprints.

    Computes a molecular fingerprint for the query and searches a local
    compound database for molecules with Tanimoto similarity above the
    threshold.

    Args:
        smiles: Query molecule in SMILES notation.
        threshold: Minimum Tanimoto similarity (0.0 to 1.0, default 0.7).
                   Values above 0.85 typically indicate the same scaffold.
        fingerprint: Fingerprint algorithm. 'morgan' (default, recommended
                     for activity similarity) or 'rdkit' (for substructure).

    Returns:
        JSON with similar compounds sorted by decreasing similarity.
    """
    from rdkit import Chem, DataStructs
    from rdkit.Chem import AllChem, RDKFingerprint

    mol = Chem.MolFromSmiles(smiles)
    if mol is None:
        return json.dumps({"error": f"Invalid SMILES: '{smiles}'"})

    # Generate fingerprint for query
    if fingerprint == "morgan":
        query_fp = AllChem.GetMorganFingerprintAsBitVect(mol, radius=2, nBits=2048)
    else:
        query_fp = RDKFingerprint(mol)

    # In production, search against a real database
    # Here we demonstrate the pattern with a small reference set
    reference_smiles = [
        ("aspirin", "CC(=O)OC1=CC=CC=C1C(=O)O"),
        ("ibuprofen", "CC(C)CC1=CC=C(C=C1)C(C)C(=O)O"),
        ("acetaminophen", "CC(=O)NC1=CC=C(O)C=C1"),
        ("naproxen", "COC1=CC2=CC(=CC2=CC1)C(C)C(=O)O"),
        ("caffeine", "CN1C=NC2=C1C(=O)N(C(=O)N2C)C"),
    ]

    results = []
    for name, ref_smi in reference_smiles:
        ref_mol = Chem.MolFromSmiles(ref_smi)
        if ref_mol is None:
            continue
        if fingerprint == "morgan":
            ref_fp = AllChem.GetMorganFingerprintAsBitVect(ref_mol, radius=2, nBits=2048)
        else:
            ref_fp = RDKFingerprint(ref_mol)

        similarity = DataStructs.TanimotoSimilarity(query_fp, ref_fp)
        if similarity >= threshold:
            results.append({
                "name": name,
                "smiles": ref_smi,
                "tanimoto_similarity": round(similarity, 4),
            })

    results.sort(key=lambda x: x["tanimoto_similarity"], reverse=True)

    return json.dumps({
        "query_smiles": smiles,
        "threshold": threshold,
        "fingerprint_type": fingerprint,
        "similar_compounds": results,
        "count": len(results),
    }, indent=2)

The code above uses two concepts that deserve formal definitions before we move on: molecular fingerprints and the Tanimoto similarity coefficient. The Tanimoto similarity coefficient used in find_similar_compounds measures the overlap between two molecular fingerprints, where a molecular fingerprint is a fixed-length bit vector in which each bit indicates the presence or absence of a particular molecular substructure. For two binary fingerprints \(A\) and \(B\), the Tanimoto coefficient is:

where \(a = |A|\), \(b = |B|\), and \(c = |A \cap B|\) are the bit counts. A value of 1.0 means identical fingerprints; values above 0.85 typically indicate compounds sharing the same chemical scaffold. This metric appears again in Chapter 49 when we build chemistry-specific discovery systems.

The SDK handles protocol boilerplate, but it cannot handle the failures inherent to external API calls and domain-specific inputs; that responsibility falls on the tool author.

5. Error Handling Patterns

Scientific tools fail in predictable ways. External APIs return 429 (rate limited), 404 (compound not found), or 500 (server error). Local computation fails on invalid input (bad SMILES), missing dependencies (RDKit not installed), or resource exhaustion (out of memory on large molecules). A well-designed MCP server handles all of these gracefully.

"""Structured error handling for MCP tools."""
import json
from enum import Enum

class ErrorCode(str, Enum):
    INVALID_INPUT = "INVALID_INPUT"
    NOT_FOUND = "NOT_FOUND"
    RATE_LIMITED = "RATE_LIMITED"
    UPSTREAM_ERROR = "UPSTREAM_ERROR"
    COMPUTATION_ERROR = "COMPUTATION_ERROR"

def tool_error(code: ErrorCode, message: str, details: dict | None = None) -> str:
    """Create a structured error response for MCP tools.

    Returns JSON that the agent can parse and act on, rather than
    raising an exception that produces an opaque error message.
    """
    error = {
        "error": True,
        "code": code.value,
        "message": message,
    }
    if details:
        error["details"] = details
    return json.dumps(error, indent=2)

# Usage in a tool:
@server.tool()
async def get_compound_by_cid(cid: int) -> str:
    """Look up a PubChem compound by its CID (compound identifier).

    Args:
        cid: PubChem Compound ID (positive integer, e.g., 2244 for aspirin).
    """
    if cid <= 0:
        return tool_error(
            ErrorCode.INVALID_INPUT,
            f"CID must be a positive integer, got {cid}.",
            {"parameter": "cid", "value": cid},
        )

    await pubchem_limiter.acquire()
    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            url = f"{PUBCHEM_BASE}/compound/cid/{cid}/property/"
            url += "IUPACName,MolecularFormula,MolecularWeight,CanonicalSMILES/JSON"
            resp = await client.get(url)

            if resp.status_code == 404:
                return tool_error(
                    ErrorCode.NOT_FOUND,
                    f"No compound found with CID {cid}.",
                    {"cid": cid, "suggestion": "Verify the CID or use search_pubchem to find it."},
                )
            if resp.status_code == 429:
                return tool_error(
                    ErrorCode.RATE_LIMITED,
                    "PubChem rate limit exceeded. Please wait before retrying.",
                    {"retry_after_seconds": 5},
                )
            resp.raise_for_status()

        except httpx.TimeoutException:
            return tool_error(
                ErrorCode.UPSTREAM_ERROR,
                "PubChem API request timed out after 30 seconds.",
                {"suggestion": "Try again or use a more specific query."},
            )
        except httpx.HTTPStatusError as e:
            return tool_error(
                ErrorCode.UPSTREAM_ERROR,
                f"PubChem API returned HTTP {e.response.status_code}.",
                {"status_code": e.response.status_code},
            )

    props = resp.json()["PropertyTable"]["Properties"][0]
    return json.dumps({
        "cid": cid,
        "name": props.get("IUPACName", "unknown"),
        "formula": props.get("MolecularFormula", ""),
        "weight": props.get("MolecularWeight", 0.0),
        "smiles": props.get("CanonicalSMILES", ""),
        "source": "PubChem",
    }, indent=2)

Once individual tools handle their own errors gracefully, the next challenge is assembling them into a single server that manages shared infrastructure efficiently.

6. Combining Tools in a Single Server

A production MCP server typically combines multiple tools behind a single process. The server manages shared resources (HTTP connection pools, rate limiters, database connections) and presents a unified interface to the client. As shown in Figure 12.2, all three tool categories share a single server process while maintaining independent rate limiters for each upstream API. Here is the pattern for combining the three tool categories we built: Figure 12.2.1 illustrates Three-pattern MCP tool architecture with shared infrastructure.

"""Combined scientific MCP server with shared infrastructure."""
import asyncio
import json

import httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server

# Shared infrastructure
server = Server("science-mcp")
_http_client: httpx.AsyncClient | None = None
_pubchem_limiter = RateLimiter(rate=5.0, burst=5)
_openalex_limiter = RateLimiter(rate=10.0, burst=10)

async def get_http_client() -> httpx.AsyncClient:
    """Lazy-initialize a shared HTTP client with connection pooling
    (reusing open TCP connections across requests instead of opening a new one each time)."""
    global _http_client
    if _http_client is None or _http_client.is_closed:
        _http_client = httpx.AsyncClient(
            timeout=30.0,
            limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
            headers={"User-Agent": "science-mcp/1.0 (research-tool)"},
        )
    return _http_client

# Register all tools on the same server
@server.tool()
async def search_pubchem(query: str, search_type: str = "name", max_results: int = 5) -> str:
    """Search PubChem for chemical compounds. [full docstring as above]"""
    await _pubchem_limiter.acquire()
    client = await get_http_client()
    # ... implementation ...

@server.tool()
async def search_literature(query: str, max_results: int = 10) -> str:
    """Search OpenAlex for scholarly papers. [full docstring as above]"""
    await _openalex_limiter.acquire()
    client = await get_http_client()
    # ... implementation ...

@server.tool()
async def compute_descriptors(smiles: str) -> str:
    """Compute molecular descriptors using RDKit. [full docstring as above]"""
    # Local computation, no rate limiting needed
    # ... implementation ...

async def main():
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

Exercises

1. Conceptual: Why does the search_pubchem tool echo back the query and search_type in its response? What problems would arise if it did not, particularly when an agent chains multiple search calls in a single reasoning trace?
2. Coding: Extend the compute_descriptors tool to include Veber's rules (PSA ≤ 140 Ų and rotatable bonds ≤ 10) alongside Lipinski's Rule of Five, where Lipinski's Rule of Five is a set of four property thresholds (molecular weight ≤ 500, LogP ≤ 5, hydrogen bond donors ≤ 5, acceptors ≤ 10) predicting whether a compound is likely to be orally bioavailable, and LogP is the partition coefficient measuring how readily a molecule dissolves in fat versus water. Return a combined "oral_bioavailability" assessment that passes only if both rule sets are satisfied. Test with at least five drugs of varying sizes.
3. Analysis: The _reconstruct_abstract function in the OpenAlex tool is \(O(n \log n)\) due to sorting. For a typical abstract of 200 words, is this a meaningful bottleneck? At what abstract length would it matter? Propose an \(O(n)\) alternative and measure the practical difference using Python's timeit.