Query Your Logs with AI: Building an MCP Powered ES Log Search System

If you have ever spent 20 minutes writing a Kibana query just to find one error, this post is for you.

I built a small project on a single AWS server where I could just type “show me the last 502 errors” in plain English, and get a real answer pulled straight from Elasticsearch. No query language, no dashboard clicking around. Here is what I used, why I used it, and how I set the whole thing up.

First, what is MCP?

MCP stands for Model Context Protocol. Anthropic (the company behind Claude) built it and made it open, so anyone can use it.

Here’s the simple way to think about it: an AI model like Claude is smart, but on its own it can’t reach into your systems and check anything live. It doesn’t know what’s in your Elasticsearch, your database, or your CloudWatch dashboard. MCP is the bridge that lets it reach out and actually check.

What is Elasticsearch doing here?

Elasticsearch is just where the logs live   a search engine built to hold huge amounts of log data and let you filter through it fast. In production, it’s usually paired with Fluentd or Fluent Bit (which ship logs from your apps into it) and Kibana (a dashboard to view them)  that combo is called EFK.

For this POC, I skipped real log collection and just wrote a small script to generate fake entries  random status codes, random service names, enough to search through.

MCP tools are just normal Python functions, exposed so an AI model can call them when needed. So instead of building a chatbot from scratch to understand “show me errors,” I wrote one function that does that job, and let MCP make it available to Claude.

What does the Python code actually do?
There are two small Python scripts in this whole setup, and neither is complicated.

Script 1 — the log generator. This one just creates fake log entries and pushes them into Elasticsearch, one by one. Each entry has a timestamp, a status code, a service name, and a message. Nothing fancy, just enough to make the data feel real.

Script 2 — the MCP server. This is the actual bridge I mentioned earlier. It’s a Python file that defines a handful of functions, each doing one specific job:

• one function searches logs by status code (so you can ask “show me 502s”)
• one counts how many logs match a status code
• one searches logs by service name
• one groups up all the errors by which service they came from

Each function talks to Elasticsearch using a normal HTTP request, gets the result back as JSON, and turns it into a plain readable string. That’s it. The only “magic” part is one line at the top of each function  a small decorator, @mcp.tool()  that tells MCP “hey, this function is something an AI model is allowed to call.”

Mcp server code

And what does Claude actually do?
Claude is the one reading your question and deciding what to do with it.

When you type something like “how many logs have status 402,” Claude reads that, looks at the list of tools available to it (the ones from our MCP server), and picks the one that matches  in this case, the “count by status” function. It runs that function, gets the number back, and replies to you in a normal sentence instead of just spitting out raw numbers.

So to break down the roles simply:

Elasticsearch — stores and searches the log data
Python MCP server — acts as  bridge between claude and Elasticsearch’s translating tool request into ES queries
Claude — reads your question, decides which MCP tool to use, and explains the answer back to you in plain english
None of these three pieces know how to do the whole job alone. Put together, they let you “talk” to your logs.

Setting it all up, step by step

Include a more in-depth explanation of the topic.

1. Launched an EC2 instance Used Amazon Linux. First tried the smallest, cheapest option (t3.micro), but it didn’t have enough memory to even install Elasticsearch — the install process got killed halfway. Moved up to a slightly bigger instance (c7i-flex.large, 4GB RAM) and that fixed it.

2. Installed Elasticsearch on the same server Added Elastic’s official package repo and installed it with a single command. It came with security turned on by default, so I had to set a password for it before I could run any real queries.

sudo rpm –import https://artifacts.elastic.co/GPG-KEY-elasticsearch

sudo tee /etc/yum.repos.d/elasticsearch.repo << ‘EOF’
[elasticsearch]
name=Elasticsearch repository for 8.x packages
baseurl=https://artifacts.elastic.co/packages/8.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
EOF

sudo dnf install -y elasticsearch
sudo systemctl start elasticsearch
sudo systemctl enable elasticsearch

/usr/share/elasticsearch/bin/elasticsearch-reset-password -u elastic

3. Installed a newer Python version The default Python on the server was too old for the MCP library, so I installed Python 3.11 alongside it.

sudo yum install -y python3.11 python3.11-pip

4. Wrote the log generator script This created 500 fake log entries with random status codes and services, and pushed them straight into Elasticsearch.

cat > ~/seed_logs.py << ‘EOF’
import requests, random, urllib3
from datetime import datetime, timedelta

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

ES = “https://localhost:9200”
INDEX = “app-logs”
AUTH = (“elastic”, “YOUR_PASSWORD_HERE”)

statuses = [200, 200, 200, 301, 404, 402, 500, 502, 503]
services = [“payment”, “auth”, “checkout”, “inventory”, “search”]

for i in range(500):
doc = {
“@timestamp”: (datetime.utcnow() – timedelta(minutes=random.randint(0, 1440))).isoformat(),
“status”: random.choice(statuses),
“service”: random.choice(services),
“message”: f”Request handled with status {random.choice(statuses)}”,
“level”: random.choice([“INFO”, “WARN”, “ERROR”]),
}
requests.post(f”{ES}/{INDEX}/_doc”, json=doc, auth=AUTH, verify=False)

print(“Seeded 500 sample log documents.”)
EOF

sudo python3.11 -m pip install requests
python3.11 ~/seed_logs.py
(Replace YOUR_PASSWORD_HERE with the password from Step 2.)

5. Installed the MCP library One pip install, and I had everything needed to build the server.

  sudo python3.11 -m pip install “mcp<2”

6. Wrote the MCP server This is the file with the four tools I mentioned above — search by status, count by status, search by service, and an error summary.

cat > ~/es_mcp_server.py << ‘EOF’
from mcp.server.fastmcp import FastMCP
import requests, urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

mcp = FastMCP(“es-log-search”)

ES = “https://localhost:9200”
INDEX = “app-logs”
AUTH = (“elastic”, “YOUR_PASSWORD_HERE”)

@mcp.tool()
def search_by_status(status_code: int, limit: int = 10) -> str:
“””Search log entries by HTTP status code (e.g. 404, 500, 502).”””
query = {“size”: limit, “sort”: [{“@timestamp”: “desc”}], “query”: {“match”: {“status”: status_code}}}
resp = requests.post(f”{ES}/{INDEX}/_search”, json=query, auth=AUTH, verify=False).json()
hits = resp.get(“hits”, {}).get(“hits”, [])
if not hits:
return f”No logs found with status {status_code}.”
return “n”.join(f”- [{h[‘_source’][‘@timestamp’]}] {h[‘_source’][‘service’]} — status {h[‘_source’][‘status’]}” for h in hits)

@mcp.tool()
def count_by_status(status_code: int) -> str:
“””Count how many log entries have a given HTTP status code.”””
query = {“query”: {“match”: {“status”: status_code}}}
resp = requests.post(f”{ES}/{INDEX}/_count”, json=query, auth=AUTH, verify=False).json()
return f”Found {resp.get(‘count’, 0)} logs with status {status_code}.”

@mcp.tool()
def search_by_service(service: str, limit: int = 10) -> str:
“””Search recent log entries for a specific service (e.g. payment, auth, checkout).”””
query = {“size”: limit, “sort”: [{“@timestamp”: “desc”}], “query”: {“match”: {“service”: service}}}
resp = requests.post(f”{ES}/{INDEX}/_search”, json=query, auth=AUTH, verify=False).json()
hits = resp.get(“hits”, {}).get(“hits”, [])
if not hits:
return f”No logs found for service ‘{service}’.”
return “n”.join(f”- [{h[‘_source’][‘@timestamp’]}] status {h[‘_source’][‘status’]} — {h[‘_source’][‘message’]}” for h in hits)

@mcp.tool()
def error_summary() -> str:
“””Get a summary count of all ERROR level logs grouped by service.”””
query = {“size”: 0, “query”: {“match”: {“level”: “ERROR”}}, “aggs”: {“by_service”: {“terms”: {“field”: “service.keyword”}}}}
resp = requests.post(f”{ES}/{INDEX}/_search”, json=query, auth=AUTH, verify=False).json()
buckets = resp.get(“aggregations”, {}).get(“by_service”, {}).get(“buckets”, [])
if not buckets:
return “No ERROR logs found.”
return “n”.join(f”- {b[‘key’]}: {b[‘doc_count’]} errors” for b in buckets)

if __name__ == “__main__”:
mcp.run(transport=”stdio”)
EOF

7. Installed Node.js and Claude Code Claude Code is the tool that actually lets you chat with Claude from your terminal, and it knows how to talk MCP.

sudo yum install -y nodejs npm
npm install -g @anthropic-ai/claude-code

8. Logged in and connected the MCP server One command registers the server (claude mcp add), and another (claude mcp list) confirms it’s connected properly.

Connected the MCP server

9. Started asking questions Opened Claude Code and just typed things like “show me logs with status 502” or “give me an error summary by service.” Claude picked the right tool each time and gave back a clean answer.