MCP Explained: How AI Agents Connect Safely to Tools and Data

The Model Context Protocol (MCP) is an open standard that lets AI applications discover and use external tools, data, and prompts through one client–server interface. For an e-commerce assistant, an MCP server is the controlled boundary between the model and systems such as the product catalog, inventory, and notifications. That boundary is only as safe as the server makes it, and this article shows what that takes. The code targets the current MCP Python SDK (2.x) and the 2026-07-28 protocol revision, and was tested in-process with the SDK’s own client.

Problem statement

A shopper asks: “Do you have the Aurora Running Jacket in medium, and can you notify me when it is back in stock?” Without MCP, every AI product needs its own catalog, inventory, and notification integration. Authentication, validation, and audit logging then differ between them, models tend to get broader access than they need, and each new capability means changes in every product. MCP puts a standard server in front of those systems, exposing a small set of governed capabilities that any MCP-capable host can discover and call.

Conceptual explanation

MCP has three roles. The host is the AI application (a shopping assistant, support console, or agent runtime). Inside it, an MCP client connects to each MCP server, which exposes capabilities. Here, one “storefront-assistant” server exposes search_products, check_product_availability, and create_back_in_stock_alert.

Tools now declare an output schema, so results arrive as validated structured data, and carry annotations (read-only, destructive, idempotent, open-world) that hosts use to decide when to ask the user before a call. Servers can also elicit input from the human through the host, such as a confirmation. In the 2026-07-28 revision that happens through multi-round-trip requests rather than server-initiated calls. That revision also removed the initialize handshake and protocol-level sessions, and deprecated sampling, roots, and protocol logging.

The model never gets a database connection just because MCP is in use: it asks the host to call a tool, and the server validates and executes. MCP does not make a deployment secure on its own. For remote servers the specification defines OAuth-based authorization for Streamable HTTP; local stdio servers rely on the credentials of the user who launched them. Business rules, consent, and authorization of each action remain the server’s job.

Runnable Python code

The earlier version of this example pinned the SDK below version 2 and used FastMCP. That still works on the maintained 1.x line, but SDK 2.0 (July 2026) renamed it MCPServer and implements the stateless 2026-07-28 protocol, so new servers should start on 2.x. More important were the safety gaps in the tools themselves:

Identity came from the model. The alert tool took customer_email as an argument, so the model could subscribe any address. The shopper now comes from the OAuth token subject over HTTP, or from host configuration for a local single-user stdio demo; the tool has no email parameter, and an injected one is ignored.

No consent, no idempotency, no limits. Every call appended a new alert. Now the human confirms through elicitation, a (shopper, SKU, size) key makes repeats return the same alert under a lock, alerts are capped per shopper, and in-stock items are refused.

Over-sharing and weak typing. Exact inventory counts went to the model even though the article advised against exposing internal inventory, prices were binary floats, and errors were ordinary results the host could not tell from success. Now availability is a status (in, low, or out of stock), prices are decimal strings, inputs are pattern- and range-checked, failures raise ToolError (an is_error result), and every tool declares its annotations and output schema.

Stdio hygiene. stdout carries the protocol on stdio, so logs go to stderr, including an audit line for every alert.

Install

# MCP Python SDK 2.x (MCPServer); v1's FastMCP lives on only with "mcp<2"
python -m venv .venv && .venv/bin/pip install "mcp>=2.3,<3"

storefront_mcp_server.py

"""storefront_mcp_server.py - MCP server for a storefront assistant."""
import logging
import os
import sys
import threading
import time
from decimal import Decimal
from typing import Annotated, Literal

from mcp.server.auth.middleware.auth_context import get_access_token
from mcp.server.mcpserver import (AcceptedElicitation, Elicit,
                                  ElicitationResult, MCPServer, Resolve)
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import ToolAnnotations
from pydantic import BaseModel, Field

# stdout carries the protocol on stdio: log to stderr only
logging.basicConfig(stream=sys.stderr, level=logging.INFO)
audit = logging.getLogger("storefront.audit")
mcp = MCPServer("storefront-assistant")

# Replace with real catalog, inventory, and notification services
PRODUCTS = {
    "JACKET-AURORA": {"name": "Aurora Running Jacket", "brand": "NorthPeak",
                      "price": Decimal("129.00"), "currency": "USD",
                      "stock": {"S": 3, "M": 0, "L": 8}},
    "SHOE-TRAIL-01": {"name": "Trail Runner Shoes", "brand": "NorthPeak",
                      "price": Decimal("119.00"), "currency": "USD",
                      "stock": {"8": 5, "9": 2, "10": 0}},
}
ALERTS: dict[tuple[str, str, str], str] = {}   # real system: DB unique key
_alerts_lock = threading.Lock()
MAX_ALERTS_PER_SHOPPER = 10

Sku = Annotated[str, Field(pattern=r"^[A-Z0-9-]{3,32}$")]
Size = Annotated[str, Field(pattern=r"^[A-Z0-9]{1,4}$")]

class ProductSummary(BaseModel):
    sku: str
    name: str
    brand: str
    price: str            # decimal string: never binary floats for money
    currency: str

class SearchResult(BaseModel):
    results: list[ProductSummary]

class Availability(BaseModel):
    sku: str
    size: str
    status: Literal["in_stock", "low_stock", "out_of_stock"]  # no raw counts

class AlertResult(BaseModel):
    alert_id: str
    created: bool         # False = an existing alert was returned
    note: str

def current_shopper() -> str:
    """Identity from trusted context, never from model-supplied arguments:
    the OAuth token subject over HTTP, or the host-configured local user
    on stdio (a single-user demo)."""
    token = get_access_token()
    if token is not None and token.subject:
        return token.subject
    shopper = os.environ.get("STOREFRONT_SHOPPER_ID")
    if not shopper:
        raise ToolError("no authenticated shopper; cannot create alerts")
    return shopper

def _product(sku: str) -> dict:
    product = PRODUCTS.get(sku)
    if product is None:
        raise ToolError(f"unknown SKU {sku}")
    return product

@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=False))
def search_products(query: Annotated[str, Field(min_length=2, max_length=80)],
                    limit: Annotated[int, Field(ge=1, le=20)] = 5
                    ) -> SearchResult:
    """Search the catalog by product name, brand, or SKU."""
    q = query.lower()
    hits = [ProductSummary(sku=sku, name=p["name"], brand=p["brand"],
                           price=f"{p['price']:.2f}", currency=p["currency"])
            for sku, p in sorted(PRODUCTS.items())
            if q in p["name"].lower() or q in p["brand"].lower()
            or q in sku.lower()]
    return SearchResult(results=hits[:limit])

@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=False))
def check_product_availability(sku: Sku, size: Size) -> Availability:
    """Check whether a product SKU is available in a size."""
    stock = _product(sku)["stock"]
    if size not in stock:
        raise ToolError(f"size {size} is not offered for {sku}; "
                        f"offered sizes: {', '.join(stock)}")
    units = stock[size]
    status = ("out_of_stock" if units == 0
              else "low_stock" if units <= 3 else "in_stock")
    return Availability(sku=sku, size=size, status=status)

class AlertConsent(BaseModel):
    confirm: bool = Field(description="Email me when this item is back")

def ask_consent(sku: str, size: str) -> Elicit[AlertConsent]:
    # Asked of the human through the host, not decided by the model
    return Elicit(f"Notify you by email when {sku} in size {size} is back "
                  f"in stock?", AlertConsent)

@mcp.tool(annotations=ToolAnnotations(readOnlyHint=False,
                                      destructiveHint=False,
                                      idempotentHint=True,
                                      openWorldHint=False))
def create_back_in_stock_alert(
        sku: Sku, size: Size,
        consent: Annotated[ElicitationResult[AlertConsent],
                           Resolve(ask_consent)]) -> AlertResult:
    """Ask the shopper to confirm, then register a back-in-stock alert for
    the signed-in shopper. Repeated calls return the same alert."""
    stock = _product(sku)["stock"]
    if size not in stock:
        raise ToolError(f"size {size} is not offered for {sku}")
    if stock[size] > 0:
        raise ToolError("item is in stock; no alert needed")
    if not (isinstance(consent, AcceptedElicitation) and consent.data.confirm):
        raise ToolError("shopper did not confirm; no alert created")
    shopper = current_shopper()
    key = (shopper, sku, size)
    with _alerts_lock:                 # check-and-insert must be atomic
        existing = ALERTS.get(key)
        if existing is None:
            if sum(k[0] == shopper for k in ALERTS) >= MAX_ALERTS_PER_SHOPPER:
                raise ToolError("alert limit reached for this shopper")
            existing = ALERTS[key] = f"alert-{len(ALERTS) + 1}"
            created = True
        else:
            created = False
    audit.info("alert shopper=%s sku=%s size=%s id=%s created=%s ts=%d",
               shopper, sku, size, existing, created, time.time())
    return AlertResult(alert_id=existing, created=created,
                       note="Demo: stored in memory; no email was sent.")

if __name__ == "__main__":
    mcp.run()   # stdio; for remote use Streamable HTTP with OAuth

Host configuration

Formats differ between hosts; this is the common mcpServers shape for a local stdio server. Use absolute paths, because hosts do not launch servers from your project directory. The environment variable identifies the local user for this single-user demo only — a shared or remote deployment must authenticate each shopper with OAuth over Streamable HTTP.

{
  "mcpServers": {
    "storefront-assistant": {
      "command": "/absolute/path/to/.venv/bin/python",
      "args": ["/absolute/path/to/storefront_mcp_server.py"],
      "env": {"STOREFRONT_SHOPPER_ID": "cust-1042"}
    }
  }
}

What the tests checked

Example interaction

The agent calls search_products(query="Aurora Running Jacket") and gets SKU JACKET-AURORA, then check_product_availability(sku="JACKET-AURORA", size="M") and learns it is out of stock. It calls create_back_in_stock_alert(sku="JACKET-AURORA", size="M"); the server asks the shopper, through the host, to confirm email notification, then registers the alert for the signed-in shopper and returns its ID. The agent replies that medium is out of stock and the alert is set — in this demo, adding that no email is actually sent.

Small real-world use case

An online retailer exposes a narrowly scoped MCP server to its shopping assistant: product search, availability by size, back-in-stock alerts, a size guide as a resource, and similar-product recommendations when an item is unavailable. The assistant can check availability and register follow-ups without direct database access, while the server enforces identity, consent, validation, limits, and audit. Over the network, that means Streamable HTTP with OAuth, TLS, and the operational controls below.

Common mistakes

Broad tools such as query_catalog(sql) or call_inventory_api(endpoint, payload): they defeat authorization and auditing. Prefer task-specific tools.

Taking identity from tool arguments. Emails, customer IDs, and roles supplied by the model are untrusted; derive them from the authenticated context.

Writes without consent or idempotency. Confirm with the human, key writes on (customer, item, variant), and make the check-and-insert atomic.

Trusting model-generated values. Validate formats, enumerations, and limits in the schema, and return a clear error the model can act on.

Returning too much. Exact stock, supplier costs, internal locations, and customer records do not belong in tool results.

Uncontrolled notifications and loops. Rate-limit writes and messages, cap tool calls per turn, and alert on unusual sequences.

Printing to stdout in a stdio server, which corrupts the protocol stream.

Production checklist

Expose only the capabilities the assistant needs, with accurate annotations so hosts prompt before writes. Authenticate every remote request with OAuth over Streamable HTTP and TLS, and authorize every tool call server-side against identity, consent, store access, and policy. Validate inputs with schemas and return structured, minimal outputs. Make writes idempotent with a unique key, require explicit consent for notifications, and honour communication preferences. Rate-limit sensitive tools and cap calls per conversation. Log tool name, arguments, authorization decision, result, latency, and a correlation ID (with OpenTelemetry trace context where available), to stderr or your logging system. Return structured errors. Test invalid inputs, unavailable sizes, duplicates, cross-customer access, injected arguments, and notification abuse. Monitor failures, unexpected tool sequences, and write volumes. Version tool schemas like APIs, track protocol and SDK versions, and start narrow — search, availability, and alerts — before purchasing, refunds, or account changes.

Takeaway: MCP turns one-off integrations into reusable, governed capabilities. The model chooses tools; the server decides who is asking, what they may do, and whether the human agreed.

Back to all blogs