Backend Engineering · AI Infrastructure · Hands-On Tutorial

MCPElasticsearchPythonProductionAI Agents

Python 3.11+ · FastMCP · Elasticsearch

In this article

If you've spent any time building AI-powered features in the last two years, you know the tax: every new data source means a new custom integration. A bespoke retrieval layer for the orders database. Another one for the support ticket system. Another for your metrics store. Each one hand-wired to a specific LLM with a specific prompt template.

Model Context Protocol (MCP) is the answer to that tax. It's an open standard — think REST or GraphQL, but for AI agent-to-data-source communication — that decouples your data layer from any specific model. Build the MCP server once, and every MCP-compatible client (Claude, Cursor, VS Code Copilot, your own agent) can talk to it.

This post is for backend engineers who want to move past "hello world" demos and build something genuinely production-ready: with authentication, structured error handling, observability, and a clean deployment story.

We'll build an MCP server on top of Elasticsearch — a common production use case — but the patterns apply to any data source.

Why MCP exists — and why you should care

Before MCP, connecting an LLM to your data meant one of two things: stuffing everything into the context window (expensive, brittle, and limited), or building a RAG pipeline with a custom retrieval API that only your integration knows how to call.

Neither scales well. The custom API approach in particular creates a fragmented ecosystem where every team builds a slightly different integration pattern, every model needs a slightly different prompt, and there's no shared abstraction to reason about.

"MCP is to AI agents what REST was to web services: a shared contract that lets producers and consumers evolve independently."

From a backend engineering perspective, MCP gives you something concrete: a well-defined protocol (JSON-RPC 2.0 over HTTPS or stdio) for exposing your data and capabilities to any agent. You define the interface once. The agent figures out how to use it.

The practical benefits for your team:

One server surfaces your data to Claude Desktop, Cursor, your internal chatbot, and any future agent — simultaneously
Model upgrades don't require touching your data layer
You own the security boundary — agents only see what the MCP server exposes
Standardised tool definitions make it straightforward to test and audit what agents can do

How MCP actually works under the hood

MCP has a simple client-server architecture. The client is the AI application (Claude Desktop, an agent framework, your chatbot). The server is the program you write — the one that knows how to talk to your actual data.

MCP Architecture

MCP Client

Claude / Cursor / Agent

⟷

JSON-RPC 2.0
stdio / HTTP+SSE

MCP Server

Your code (FastMCP)

⟷

Internal APIs
DB / REST

Data Sources

Elasticsearch / DB / APIs

Communication uses JSON-RPC 2.0. There are two transport modes you'll care about:

stdio — the server runs as a subprocess and communicates over standard input/output. Used by Claude Desktop and most local dev setups. Zero networking, zero auth complexity.
HTTP + SSE (Server-Sent Events) — the server runs as an HTTP service. Used for remote/deployed servers. This is what you want for production.

The lifecycle is straightforward: client connects and sends an initialize request, server responds with its capabilities (what tools, resources, and prompts it exposes), then the client can start making tool calls. The server processes them, hits your actual data sources, and returns structured results.

Protocol detail

MCP is transport-agnostic at the spec level. The JSON-RPC messages are identical whether you're running over stdio or HTTP. This means the same server code can support both transports with minimal changes — useful for local development (stdio) vs. production (HTTP).

The three MCP primitives

Everything in MCP is expressed through three primitives. Understanding these is the key to designing a good MCP server.

Primitive	What it is	When to use it	Read/Write
Resources	Structured data the model can read as context	Documents, records, search results, config	Read-only
Tools	Executable functions the agent can invoke	Search, create, update, call external APIs	Read + Write
Prompts	Reusable prompt templates with parameters	Standardising common workflows, slash commands	Read-only

For most backend use cases, Tools are where you spend 80% of your time. They're the functions agents actually call — search an index, fetch a record, trigger a workflow. Design them like well-typed REST endpoints: explicit inputs, explicit outputs, explicit errors.

Project setup and dependencies

We'll use FastMCP — a Python framework that wraps the MCP protocol and lets you define tools with plain decorated functions. Think of it as FastAPI, but for agent-facing APIs.

bash

# Create project structure
mkdir mcp-es-server && cd mcp-es-server
python -m venv .venv && source .venv/bin/activate

# Core dependencies
pip install fastmcp elasticsearch[async] pydantic python-dotenv

# Dev/observability
pip install pytest pytest-asyncio httpx opentelemetry-sdk

Project layout:

text

mcp-es-server/
├── src/
│   ├── server.py          # FastMCP app + tool registration
│   ├── config.py          # Settings via pydantic-settings
│   ├── es_client.py       # Elasticsearch client wrapper
│   ├── tools/
│   │   ├── search.py      # Search tools
│   │   ├── index_info.py  # Index introspection tools
│   │   └── write.py       # Write/update tools
│   └── middleware/
│       ├── auth.py        # API key validation
│       └── observability.py
├── tests/
├── Dockerfile
├── docker-compose.yml
└── .env.example

Building the server step by step

Step 1Configuration with Pydantic Settings

Hardcoded values are a production anti-pattern. Use pydantic-settings to load config from environment variables with type validation.

python — src/config.py

from pydantic_settings import BaseSettings
from pydantic import AnyUrl, SecretStr
from functools import lru_cache


class Settings(BaseSettings):
    # Elasticsearch
    es_url: AnyUrl = "http://localhost:9200"
    es_api_key: SecretStr | None = None
    es_username: str | None = None
    es_password: SecretStr | None = None
    es_timeout: int = 30
    es_max_results: int = 50

    # MCP server
    mcp_api_key: SecretStr | None = None
    mcp_transport: str = "stdio"        # "stdio" | "streamable-http"
    mcp_host: str = "0.0.0.0"
    mcp_port: int = 8000

    # Observability
    log_level: str = "INFO"
    otel_endpoint: str | None = None

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"


@lru_cache
def get_settings() -> Settings:
    return Settings()

Step 2Elasticsearch Client Wrapper

Wrap the official Elasticsearch async client with connection pooling and a clean interface your tools can depend on.

python — src/es_client.py

from elasticsearch import AsyncElasticsearch, NotFoundError
from contextlib import asynccontextmanager
from .config import get_settings
import logging

logger = logging.getLogger(__name__)
_client: AsyncElasticsearch | None = None


async def get_client() -> AsyncElasticsearch:
    global _client
    if _client is None:
        settings = get_settings()
        kwargs = {
            "hosts": [str(settings.es_url)],
            "request_timeout": settings.es_timeout,
            "retry_on_timeout": True,
            "max_retries": 3,
        }
        if settings.es_api_key:
            kwargs["api_key"] = settings.es_api_key.get_secret_value()
        elif settings.es_username:
            kwargs["basic_auth"] = (
                settings.es_username,
                settings.es_password.get_secret_value()
            )
        _client = AsyncElasticsearch(**kwargs)
        logger.info("Elasticsearch client initialised")
    return _client


async def close_client():
    global _client
    if _client:
        await _client.close()
        _client = None

Step 3Defining Tools with FastMCP

This is the core of your server. Each @mcp.tool() decorated function becomes a callable that any MCP-compatible agent can discover and invoke. Type annotations are automatically converted to JSON Schema — so what you write is what agents see.

python — src/tools/search.py

from fastmcp import FastMCP
from pydantic import BaseModel, Field
from typing import Annotated
from ..es_client import get_client
from ..config import get_settings
from .errors import MCPToolError


class SearchResult(BaseModel):
    index: str
    id: str
    score: float
    source: dict
    highlight: dict | None = None


def register_search_tools(mcp: FastMCP):

    @mcp.tool()
    async def search(
        index: Annotated[str, Field(description="Elasticsearch index name or pattern")],
        query: Annotated[str, Field(description="Natural language search query")],
        size: Annotated[int, Field(
            description="Max results to return", ge=1, le=100
        )] = 10,
        fields: Annotated[list[str] | None, Field(
            description="Specific fields to return. None returns all."
        )] = None,
    ) -> list[SearchResult]:
        """
        Full-text search across an Elasticsearch index.
        Returns ranked results with optional field highlighting.
        """
        settings = get_settings()
        size = min(size, settings.es_max_results)

        try:
            es = await get_client()
            resp = await es.search(
                index=index,
                body={
                    "query": {
                        "multi_match": {
                            "query": query,
                            "type": "best_fields",
                            "fuzziness": "AUTO",
                        }
                    },
                    "highlight": {
                        "fields": {"*": {"fragment_size": 150}}
                    },
                    "_source": fields or True,
                    "size": size,
                }
            )

            return [
                SearchResult(
                    index=hit["_index"],
                    id=hit["_id"],
                    score=hit["_score"],
                    source=hit["_source"],
                    highlight=hit.get("highlight"),
                )
                for hit in resp["hits"]["hits"]
            ]

        except Exception as e:
            raise MCPToolError(
                tool="search",
                message=str(e),
                recoverable=True,
            ) from e

    @mcp.tool()
    async def list_indices(
        pattern: Annotated[str, Field(
            description="Index pattern to filter (e.g. 'logs-*'). Default: all indices."
        )] = "*",
    ) -> list[dict]:
        """List all Elasticsearch indices matching the given pattern."""
        es = await get_client()
        indices = await es.cat.indices(
            index=pattern, h="index,docs.count,store.size,health", format="json"
        )
        return indices

    @mcp.tool()
    async def get_mappings(
        index: Annotated[str, Field(description="Index name")],
    ) -> dict:
        """
        Retrieve field mappings for an index.
        Useful for understanding the schema before constructing a search.
        """
        es = await get_client()
        result = await es.indices.get_mapping(index=index)
        return dict(result)

Step 4The Main Server Entry Point

python — src/server.py

from fastmcp import FastMCP
from contextlib import asynccontextmanager
from .config import get_settings
from .es_client import get_client, close_client
from .tools.search import register_search_tools
from .tools.index_info import register_index_tools
from .tools.write import register_write_tools
from .middleware.auth import require_api_key
import logging

logging.basicConfig(level=get_settings().log_level)
logger = logging.getLogger(__name__)


@asynccontextmanager
async def lifespan(app):
    """Startup: warm the ES client. Shutdown: close connections."""
    logger.info("MCP server starting up...")
    await get_client()       # warm connection pool
    yield
    await close_client()     # clean shutdown
    logger.info("MCP server shut down cleanly.")


mcp = FastMCP(
    name="elasticsearch-mcp-server",
    instructions="""
    You are connected to an Elasticsearch cluster.
    Use list_indices to discover available data.
    Use get_mappings to understand schema before searching.
    Use search for full-text queries across indices.
    Always prefer targeted index searches over wildcard queries in production.
    """,
    lifespan=lifespan,
)

# Register all tool groups
register_search_tools(mcp)
register_index_tools(mcp)
register_write_tools(mcp)


if __name__ == "__main__":
    settings = get_settings()
    mcp.run(
        transport=settings.mcp_transport,
        host=settings.mcp_host,
        port=settings.mcp_port,
    )

Authentication and security

If your MCP server runs over HTTP (production), you need to authenticate incoming requests. The standard pattern is bearer token validation — a shared secret the client passes in the Authorization header.

python — src/middleware/auth.py

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse
from ..config import get_settings
import secrets


class APIKeyMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        settings = get_settings()

        # Skip auth if no key is configured (e.g. local stdio mode)
        if not settings.mcp_api_key:
            return await call_next(request)

        auth_header = request.headers.get("Authorization", "")
        if not auth_header.startswith("Bearer "):
            return JSONResponse(
                {"error": "Missing or malformed Authorization header"},
                status_code=401
            )

        token = auth_header[7:]
        expected = settings.mcp_api_key.get_secret_value()

        # Constant-time comparison to prevent timing attacks
        if not secrets.compare_digest(token, expected):
            return JSONResponse(
                {"error": "Invalid API key"},
                status_code=403
            )

        return await call_next(request)

Security note

Always set an expiration on API keys. Use short durations (1–7 days) in development, 30–90 days in production, with automated rotation. In production, restrict the Elasticsearch API key's index access to only the indices your tools need — principle of least privilege.

Additional security considerations

Index allowlisting — accept an allowed_indices config list and validate every tool call against it before hitting Elasticsearch
Query depth limiting — reject or cap deeply nested DSL queries that could cause cluster strain
Result scrubbing — strip PII fields from search results before returning them to the agent
Rate limiting — add per-client rate limits, especially if the server is accessible to multiple agents

Error handling and observability

Production MCP servers need structured errors — not bare Python exceptions that surface as cryptic messages to the agent. Define a typed error hierarchy so agents can distinguish "this index doesn't exist" from "the cluster is down".

python — src/tools/errors.py

from pydantic import BaseModel
from enum import StrEnum


class ErrorCode(StrEnum):
    INDEX_NOT_FOUND = "INDEX_NOT_FOUND"
    QUERY_INVALID   = "QUERY_INVALID"
    CLUSTER_ERROR   = "CLUSTER_ERROR"
    AUTH_DENIED     = "AUTH_DENIED"
    RATE_LIMITED    = "RATE_LIMITED"
    UNKNOWN         = "UNKNOWN"


class MCPToolError(Exception):
    def __init__(
        self,
        tool: str,
        message: str,
        code: ErrorCode = ErrorCode.UNKNOWN,
        recoverable: bool = False,
    ):
        self.tool = tool
        self.message = message
        self.code = code
        self.recoverable = recoverable
        super().__init__(
            f"[{code}] Tool '{tool}' failed: {message}"
            f"{' (recoverable)' if recoverable else ''}"
        )


def classify_es_error(e: Exception, tool: str) -> MCPToolError:
    """Map Elasticsearch exceptions to structured MCP errors."""
    from elasticsearch import (
        NotFoundError, RequestError, ConnectionError, AuthenticationException
    )
    if isinstance(e, NotFoundError):
        return MCPToolError(tool, str(e), ErrorCode.INDEX_NOT_FOUND, recoverable=True)
    if isinstance(e, RequestError):
        return MCPToolError(tool, str(e), ErrorCode.QUERY_INVALID, recoverable=True)
    if isinstance(e, AuthenticationException):
        return MCPToolError(tool, str(e), ErrorCode.AUTH_DENIED, recoverable=False)
    if isinstance(e, ConnectionError):
        return MCPToolError(tool, str(e), ErrorCode.CLUSTER_ERROR, recoverable=True)
    return MCPToolError(tool, str(e), ErrorCode.UNKNOWN)

Structured logging

Every tool call should emit a structured log line with enough context to debug in production without reproducing the query.

python — src/middleware/observability.py

import time, logging, json
from functools import wraps

logger = logging.getLogger("mcp.tools")


def instrument(tool_name: str):
    """Decorator: logs latency, result count, and errors for every tool call."""
    def decorator(fn):
        @wraps(fn)
        async def wrapper(*args, **kwargs):
            t0 = time.perf_counter()
            try:
                result = await fn(*args, **kwargs)
                elapsed = (time.perf_counter() - t0) * 1000
                logger.info(json.dumps({
                    "tool": tool_name,
                    "status": "ok",
                    "latency_ms": round(elapsed, 2),
                    "result_count": len(result) if isinstance(result, list) else 1,
                }))
                return result
            except Exception as e:
                elapsed = (time.perf_counter() - t0) * 1000
                logger.error(json.dumps({
                    "tool": tool_name,
                    "status": "error",
                    "latency_ms": round(elapsed, 2),
                    "error": str(e),
                }))
                raise
        return wrapper
    return decorator

Containerising and deploying

dockerfile

FROM python:3.12-slim

WORKDIR /app

# Install deps first (cached layer)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY src/ ./src/

# Non-root user for security
RUN useradd -m mcpuser
USER mcpuser

ENV MCP_TRANSPORT=streamable-http
ENV MCP_HOST=0.0.0.0
ENV MCP_PORT=8000

EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s \
  CMD curl -f http://localhost:8000/health || exit 1

ENTRYPOINT ["python", "-m", "src.server"]

yaml — docker-compose.yml

services:
  mcp-server:
    build: .
    ports:
      - "8000:8000"
    env_file: .env
    environment:
      - MCP_TRANSPORT=streamable-http
    restart: unless-stopped
    depends_on:
      elasticsearch:
        condition: service_healthy

  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.13.0
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
    ports:
      - "9200:9200"
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9200"]
      interval: 10s
      timeout: 5s
      retries: 5

Connecting Claude Desktop to your server

For local development using stdio, add this to your Claude Desktop config (~/Library/Application Support/Claude/claude_desktop_config.json on macOS):

json — claude_desktop_config.json

{
  "mcpServers": {
    "elasticsearch": {
      "command": "python",
      "args": ["-m", "src.server"],
      "cwd": "/path/to/mcp-es-server",
      "env": {
        "ES_URL": "http://localhost:9200",
        "ES_API_KEY": "your-api-key-here",
        "MCP_TRANSPORT": "stdio"
      }
    }
  }
}

Testing your MCP server

MCP servers are async Python — test them like you'd test any async service. Use pytest-asyncio and mock the Elasticsearch client.

python — tests/test_search.py

import pytest
from unittest.mock import AsyncMock, patch
from src.server import mcp


@pytest.mark.asyncio
async def test_search_returns_results():
    mock_response = {
        "hits": {"hits": [{
            "_index": "orders",
            "_id": "1",
            "_score": 1.5,
            "_source": {"product": "Widget A", "amount": 699},
        }]}
    }

    with patch("src.es_client.get_client") as mock_get_client:
        mock_es = AsyncMock()
        mock_es.search.return_value = mock_response
        mock_get_client.return_value = mock_es

        result = await mcp.call_tool(
            "search",
            {"index": "orders", "query": "Widget"}
        )

        assert len(result) == 1
        assert result[0].id == "1"
        assert result[0].source["product"] == "Widget A"


@pytest.mark.asyncio
async def test_search_caps_result_size():
    """Ensure the size cap from config is respected."""
    with patch("src.es_client.get_client") as mock_get_client:
        mock_es = AsyncMock()
        mock_es.search.return_value = {"hits": {"hits": []}}
        mock_get_client.return_value = mock_es

        await mcp.call_tool(
            "search",
            {"index": "orders", "query": "test", "size": 9999}
        )

        call_kwargs = mock_es.search.call_args["body"]
        assert call_kwargs["size"] <= 50  # es_max_results default

Integration testing

For integration tests, spin up a real Elasticsearch instance via docker-compose in CI. Seed a test index with known documents and verify your tools return the expected results end-to-end. This catches query DSL bugs that mocks can't.

Production readiness checklist

Before you ship an MCP server to production, work through this list. Every item has caused a production incident somewhere.

API key authentication enabled for HTTP transport
Elasticsearch credentials stored in secrets manager, not environment files in the repo
Index allowlist configured — agents can only access intended indices
Result size cap enforced in code, not just in the tool's field description
Structured JSON logging for every tool call (tool name, latency, result count, error code)
Health check endpoint returning Elasticsearch cluster status
Graceful shutdown handling — in-flight requests complete before process exits
Connection pooling configured with appropriate pool size for your concurrency level
Error taxonomy defined — agents can distinguish recoverable from non-recoverable errors
Non-root Docker user in production container image
API key expiry and rotation policy documented and automated
Unit tests cover error paths, not just happy paths
Integration test suite runs in CI against a real Elasticsearch instance
Tool descriptions are accurate — agents make decisions based on your docstrings

MCP is a straightforward protocol once you've built one server. The primitives are small, the transport is familiar (JSON-RPC over HTTP), and FastMCP handles most of the boilerplate.

What makes the difference between a demo and a production system is everything else: typed errors agents can reason about, authentication that doesn't leak credentials, structured logging you can actually debug, and a deployment that survives restarts cleanly.

The patterns in this post are applicable beyond Elasticsearch — the same structure works for Postgres, Redis, internal REST APIs, or any data source you want to expose to agents. Build it once, connect it to everything.