models with extra="forbid" reject malformed payloads before they reach business logic. 3. Stateless Transport Design: The server does not rely on protocol sessions. Any required state is passed explicitly as tool arguments. 4. Model-Readable Errors: Error responses are structured as recovery instructions, not stack traces or HTTP codes. 5. Honest Annotations: Tool metadata accurately reflects side effects, idempotency, and external dependencies to enable host-level consent flows.

Implementation

from __future__ import annotations

import os
import logging
from typing import Annotated
from pydantic import BaseModel, Field, ConfigDict
import httpx
from mcp.server.fastmcp import FastMCP

# Server identifier follows {domain}_mcp naming convention
server = FastMCP("inventory_mcp")

API_ENDPOINT = "https://api.warehouse-systems.internal/v2"
logger = logging.getLogger("mcp.inventory")


class StockQuery(BaseModel):
    """Validates and structures inventory lookup requests."""
    model_config = ConfigDict(
        str_strip_whitespace=True,
        extra="forbid",
        frozen=True,
    )

    sku: Annotated[str, Field(
        min_length=3, 
        max_length=20, 
        pattern=r"^[A-Z0-9-]+$",
        description="Stock keeping unit identifier"
    )]
    warehouse_id: Annotated[str, Field(
        min_length=1,
        max_length=50,
        description="Target warehouse location code"
    )]


def _format_recovery_message(exception: Exception) -> str:
    """Converts technical failures into actionable model instructions."""
    if isinstance(exception, httpx.HTTPStatusError):
        status = exception.response.status_code
        if status == 404:
            return "Inventory lookup failed: SKU or warehouse not found. Verify the identifier format and retry."
        if status == 429:
            return "Rate limit reached. Pause for 30 seconds before attempting another lookup."
        return f"External service returned status {status}. Check network connectivity and retry."
    if isinstance(exception, httpx.TimeoutException):
        return "Request timed out after 10 seconds. The warehouse API may be degraded. Retry with a shorter timeout."
    return f"Unexpected failure: {type(exception).__name__}. Log the request ID and escalate to infrastructure."


@server.tool(
    name="check_stock_level",
    annotations={
        "title": "Query warehouse inventory",
        "readOnlyHint": True,
        "openWorldHint": True,
        "idempotentHint": True,
    },
)
async def retrieve_stock(query: StockQuery) -> str:
    """Fetch current stock levels for a specific SKU and warehouse.

    Args:
        query: Validated SKU and warehouse identifier.
    Returns:
        Formatted inventory status or recovery instructions.
    """
    api_token = os.environ.get("WAREHOUSE_API_TOKEN")
    if not api_token:
        return "Configuration error: WAREHOUSE_API_TOKEN is not set in the runtime environment."

    request_url = f"{API_ENDPOINT}/inventory/{query.warehouse_id}/sku/{query.sku}"
    
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.get(
                request_url,
                headers={"Authorization": f"Bearer {api_token}"},
            )
            response.raise_for_status()
            payload = response.json()
    except Exception as exc:
        logger.warning("Stock lookup failed for %s in %s", query.sku, query.warehouse_id, exc_info=exc)
        return _format_recovery_message(exc)

    available = payload.get("available_units", 0)
    reserved = payload.get("reserved_units", 0)
    return (
        f"Stock status for {query.sku} at {query.warehouse_id}:\n"
        f"  Available: {available}\n"
        f"  Reserved: {reserved}\n"
        f"  Net: {available - reserved}"
    )


if __name__ == "__main__":
    server.run()

Why These Choices Matter

• extra="forbid" in Pydantic: Prevents the model from injecting unexpected fields that could bypass validation or trigger unintended code paths.
• frozen=True: Ensures the input model cannot be mutated during execution, preventing race conditions in concurrent async contexts.
• Explicit error formatting: LLMs operate on token probabilities. Vague errors increase hallucination risk. Structured recovery paths keep the model on-task.
• Annotation honesty: readOnlyHint and idempotentHint allow the host to skip confirmation dialogs for safe operations. Mislabeling breaks the trust contract and forces unnecessary user friction.
• Stateless by design: The server does not cache session state. If a workflow requires multi-step coordination, the host passes explicit identifiers (e.g., order_id) as tool arguments. This aligns with the 2026-07-28 spec and enables zero-downtime deployments.

Pitfall Guide

1. Treating MCP as Function Calling

Explanation: Function calling is a vendor-specific mechanism where the model directly invokes code within a single application. MCP is a transport and negotiation protocol that enables cross-host capability discovery. Fix: Design servers as independent services that advertise capabilities via JSON-RPC. Never assume the host will execute your code directly; always treat calls as external requests requiring validation and auth.

2. Conflating Tools, Resources, and Prompts

Explanation: Tools execute logic, resources expose read-only data under URIs, and prompts provide reusable instruction templates. Mixing them creates ambiguous contracts and breaks host routing. Fix: Use tools for side-effecting or complex operations. Use resources for static/semi-static context (e.g., documentation, configuration files). Use prompts for repeatable user workflows. Keep boundaries strict.

3. Writing Human-Centric Error Messages

Explanation: Error responses are consumed by the model, not developers. Stack traces or HTTP codes provide no recovery path and increase token waste. Fix: Format errors as actionable instructions. Include what failed, why it likely happened, and what the model should attempt next. Log technical details server-side; return semantic guidance to the host.

4. Misusing Tool Annotations

Explanation: Annotations like readOnlyHint or destructiveHint drive host consent flows. Lying about behavior breaks the security model and can trigger automatic user warnings or request rejection. Fix: Audit every tool against the annotation spec. Set destructiveHint only for irreversible operations. Use idempotentHint for safe retries. Never claim readOnlyHint on write operations.

5. Assuming Sticky Sessions

Explanation: The 2026-07-28 revision removes Mcp-Session-Id. Protocol-level state is gone. Relying on session affinity breaks horizontal scaling and causes request routing failures. Fix: Design stateless servers. Pass required context as explicit tool arguments. Use external caches or databases if cross-call state is necessary. Treat every request as independent.

6. Synchronous I/O in Async Context

Explanation: Blocking calls (e.g., requests.get, synchronous database drivers) stall the event loop. One slow request blocks all concurrent clients. Fix: Use async HTTP clients (httpx.AsyncClient, aiohttp), async database drivers, and asyncio-compatible libraries. Never mix sync and async I/O in the same execution path.

7. Hardcoding Secrets in Descriptions

Explanation: Tool descriptions and results are untrusted. Embedding API keys, tokens, or internal URLs in metadata exposes them to prompt injection or model leakage. Fix: Store secrets in environment variables or secret managers. Never include credentials in docstrings, annotations, or return payloads. Validate all external inputs before use.

Production Bundle

Action Checklist

• Validate all inputs with Pydantic models using extra="forbid" and strict type constraints
• Implement async I/O exclusively; replace all synchronous network/database calls
• Format error responses as recovery instructions, not technical logs
• Set tool annotations accurately to match actual side effects and idempotency
• Store secrets in environment variables; never embed in code or descriptions
• Design for stateless routing; pass workflow state explicitly as tool arguments
• Add structured logging with request IDs for observability and debugging
• Implement circuit breakers or retry limits for external API dependencies

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
Local CLI integration	stdio transport	Single-process, inherits OS auth, zero network overhead	Minimal infrastructure cost
Multi-client SaaS backend	Streamable HTTP	Supports concurrent clients, OAuth 2.1/OIDC auth, load balancer compatible	Moderate (requires HTTPS, auth service)
Read-only documentation	Resource primitive	Host caches URI content, reduces token usage, no execution overhead	Low (static hosting)
Complex workflow orchestration	Tool primitive	Enables validation, side effects, and explicit state passing	Higher (compute + external API costs)
High-traffic production	Stateless HTTP + async I/O	Aligns with 2026 spec, enables horizontal scaling without sticky sessions	Predictable (scales linearly with load)

Configuration Template

# mcp_server_config.py
import os
import logging
from mcp.server.fastmcp import FastMCP

# Environment-driven configuration
TRANSPORT = os.getenv("MCP_TRANSPORT", "streamable_http")
PORT = int(os.getenv("MCP_PORT", "8000"))
LOG_LEVEL = os.getenv("MCP_LOG_LEVEL", "INFO")

# Initialize server with explicit metadata
server = FastMCP(
    name="production_mcp",
    version="1.2.0",
    instructions="Stateless inventory and order management server. All tools are idempotent unless annotated otherwise."
)

# Structured logging setup
logging.basicConfig(
    level=getattr(logging, LOG_LEVEL.upper(), logging.INFO),
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    handlers=[logging.StreamHandler()]
)

# Transport routing
def start_server():
    if TRANSPORT == "stdio":
        server.run()
    elif TRANSPORT == "streamable_http":
        server.run(transport="streamable_http", port=PORT)
    else:
        raise ValueError(f"Unsupported transport: {TRANSPORT}")

if __name__ == "__main__":
    start_server()

Quick Start Guide

1. Install dependencies: pip install mcp pydantic httpx
2. Create the server file: Save the configuration template and implementation code into server.py
3. Set environment variables: export WAREHOUSE_API_TOKEN="your_token_here" and export MCP_TRANSPORT="stdio"
4. Run locally: python server.py (stdio) or python server.py with MCP_TRANSPORT=streamable_http for network access
5. Connect a host: Configure your MCP-compatible client (Claude Desktop, Cursor, or custom agent) to point to the server endpoint or stdio process

This architecture aligns with the current MCP specification, enforces strict security boundaries, and scales horizontally without protocol-level session management. Treat the server as a stateless capability provider, validate everything, and let the host manage orchestration.