Building a Log Analysis Agent with MCP (Model Context Protocol): Complete Guide

Introduction to MCP and Log Analysis Agents

The Model Context Protocol (MCP) is an open standard that enables AI assistants to securely connect to external data sources and tools. Think of it as USB-C for AI applications: a universal interface that lets language models read from databases, query APIs, inspect files, and execute operations without bespoke integrations for every provider.

A Log Analysis Agent built on MCP leverages this protocol to give an LLM direct, structured access to log files, log aggregators, and observability backends. Instead of pasting log snippets into a chat window, the agent can pull logs on demand, filter by severity, correlate timestamps, and surface anomalies — all through standardized MCP tools and resources.

In this guide, you'll build a working MCP server that exposes log-searching tools, then connect it to an MCP-compatible client so an LLM can investigate incidents conversationally.

Why MCP for Log Analysis?

Log analysis is a natural fit for LLMs: logs are text-heavy, semi-structured, and require reasoning to interpret. But traditional integrations are painful. Each log vendor (Elasticsearch, Loki, Splunk, CloudWatch) has its own SDK, auth flow, and query language. MCP solves three concrete problems:

Decoupling — The agent logic doesn't need to know whether logs live in a file, Loki, or Elasticsearch. It calls the same MCP tools regardless.
Reusability — One MCP server can serve multiple clients: Claude Desktop, a custom CLI, a Slack bot, or a web dashboard.
Security boundary — The server controls exactly which logs are exposed and which actions are permitted, keeping the LLM sandboxed.

Prerequisites and Project Setup

You'll need Python 3.10+, the official MCP SDK, and a sample log file. Create a new project directory and install dependencies:

mkdir log-mcp-agent && cd log-mcp-agent
python -m venv .venv
source .venv/bin/activate
pip install "mcp[cli]" httpx

Create a sample log file to work with:

# logs/app.log
2024-11-14T09:12:01Z INFO  auth user=alice action=login ip=10.0.0.12
2024-11-14T09:12:45Z INFO  api  route=/orders status=200 latency=42ms
2024-11-14T09:13:02Z WARN  db   query=SELECT * FROM users duration=1200ms
2024-11-14T09:13:30Z ERROR auth user=bob action=login reason=bad_password ip=10.0.0.99
2024-11-14T09:14:10Z ERROR api  route=/payments status=500 latency=8800ms
2024-11-14T09:14:11Z ERROR api  route=/payments status=500 latency=9100ms
2024-11-14T09:14:12Z ERROR api  route=/payments status=500 latency=8950ms
2024-11-14T09:15:00Z INFO  auth user=alice action=logout ip=10.0.0.12

Building the MCP Server

The MCP server exposes tools (callable functions) and resources (readable data). For log analysis, we'll expose three tools: search_logs, count_by_level, and tail_logs. We'll also expose the raw log file as a resource.

Server Implementation

Create log_server.py:

from mcp.server.fastmcp import FastMCP
from pathlib import Path
from collections import Counter
import re

LOG_PATH = Path(__file__).parent / "logs" / "app.log"

mcp = FastMCP("log-analyzer")

LOG_LINE_RE = re.compile(
    r'(?P<ts>\S+)\s+(?P<level>\w+)\s+(?P<component>\w+)\s+(?P<msg>.*)'
)

def _parse(line: str):
    m = LOG_LINE_RE.match(line.strip())
    if not m:
        return None
    return m.groupdict()

@mcp.tool()
def search_logs(level: str = "", component: str = "", keyword: str = "", limit: int = 50) -> str:
    """Search application logs by level, component, or keyword.

    Args:
        level: Filter by log level (INFO, WARN, ERROR). Empty = all.
        component: Filter by component (auth, api, db). Empty = all.
        keyword: Substring to match in the message body.
        limit: Maximum number of lines to return.
    """
    results = []
    for line in LOG_PATH.read_text().splitlines():
        parsed = _parse(line)
        if not parsed:
            continue
        if level and parsed["level"] != level.upper():
            continue
        if component and parsed["component"] != component.lower():
            continue
        if keyword and keyword.lower() not in parsed["msg"].lower():
            continue
        results.append(line)
        if len(results) >= limit:
            break
    return "\n".join(results) if results else "No matching log lines found."

@mcp.tool()
def count_by_level() -> str:
    """Return a histogram of log lines grouped by severity level."""
    counter = Counter()
    for line in LOG_PATH.read_text().splitlines():
        parsed = _parse(line)
        if parsed:
            counter[parsed["level"]] += 1
    if not counter:
        return "No logs available."
    return "\n".join(f"{lvl}: {n}" for lvl, n in counter.most_common())

@mcp.tool()
def tail_logs(n: int = 20) -> str:
    """Return the last N log lines."""
    lines = LOG_PATH.read_text().splitlines()
    return "\n".join(lines[-n:])

@mcp.resource("log://app.log")
def raw_log() -> str:
    """The complete raw application log file."""
    return LOG_PATH.read_text()

if __name__ == "__main__":
    mcp.run(transport="stdio")

The FastMCP class handles protocol negotiation, JSON-RPC framing, and tool dispatch. Each @mcp.tool() decorator registers a function whose docstring becomes the schema the LLM sees — so write docstrings carefully, as they drive tool selection.

Testing the Server Standalone

Before wiring up an LLM, verify the server works with the MCP CLI inspector:

mcp dev log_server.py

This opens a local web UI where you can call search_logs, count_by_level, and tail_logs directly and inspect the JSON responses.

Connecting a Client

Now build a client that launches the server as a subprocess and routes an LLM's tool calls to it. We'll use Anthropic's SDK for the model, but the MCP client code is provider-agnostic.

# agent_client.py
import asyncio
import json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from anthropic import Anthropic

SERVER_CMD = ["python", "log_server.py"]

async def run_agent(user_query: str):
    server_params = StdioServerParameters(command=SERVER_CMD[0], args=SERVER_CMD[1:])

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Discover available tools from the server
            tools_result = await session.list_tools()
            tool_specs = [
                {
                    "name": t.name,
                    "description": t.description,
                    "input_schema": t.inputSchema,
                }
                for t in tools_result.tools
            ]

            client = Anthropic()
            messages = [{"role": "user", "content": user_query}]

            # Agentic loop: model calls tools until it produces a final answer
            while True:
                response = client.messages.create(
                    model="claude-3-5-sonnet-20241022",
                    max_tokens=2048,
                    tools=tool_specs,
                    messages=messages,
                )
                messages.append({"role": "assistant", "content": response.content})

                if response.stop_reason == "end_turn":
                    for block in response.content:
                        if block.type == "text":
                            print(block.text)
                    break

                # Execute any tool calls the model requested
                for block in response.content:
                    if block.type == "tool_use":
                        result = await session.call_tool(
                            block.name, block.input
                        )
                        messages.append({
                            "role": "user",
                            "content": [{
                                "type": "tool_result",
                                "tool_use_id": block.id,
                                "content": result.content[0].text,
                            }],
                        })

if __name__ == "__main__":
    query = "Are there any errors in the logs? If so, summarize what happened and which component is responsible."
    asyncio.run(run_agent(query))

Run it:

export ANTHROPIC_API_KEY=sk-ant-...
python agent_client.py

The model will autonomously call count_by_level, then search_logs with level=ERROR, synthesize the results, and report that the api component is failing on the /payments route with repeated 500s.

Extending to Real Log Backends

File-based logs are fine for demos, but production agents need to query real backends. The beauty of MCP is that the client code doesn't change — only the server implementation does. Here's a Loki-backed variant of search_logs:

import httpx
from datetime import datetime, timedelta

LOKI_URL = "http://loki:3100"

@mcp.tool()
async def search_loki(query: str, minutes: int = 15, limit: int = 100) -> str:
    """Query Loki logs using LogQL.

    Args:
        query: A LogQL query string, e.g. '{job="api"} |= "ERROR"'.
        minutes: Lookback window in minutes.
        limit: Maximum number of log lines to return.
    """
    end = datetime.utcnow()
    start = end - timedelta(minutes=minutes)
    params = {
        "query": query,
        "start": int(start.timestamp() * 1e9),
        "end": int(end.timestamp() * 1e9),
        "limit": limit,
    }
    async with httpx.AsyncClient() as client:
        r = await client.get(f"{LOKI_URL}/loki/api/v1/query_range", params=params)
        r.raise_for_status()
        data = r.json()
    lines = [stream["value"] for result in data["data"]["result"]
             for stream in result["values"]]
    return "\n".join(v[1] for v in lines) if lines else "No matching logs."

Swap this in and the agent immediately gains the ability to query live infrastructure — no client changes required.

Best Practices

Keep tools narrow and composable. One tool that does "search + filter + summarize" is harder for an LLM to use correctly than three focused tools it can chain. The model is better at composing simple verbs than guessing the right parameters for a monolithic function.
Write precise docstrings. The docstring is the tool's contract with the model. Document every parameter, list valid values, and give an example. Vague docstrings cause hallucinated arguments.
Cap result sizes. LLMs choke on megabytes of log text. Enforce limit defaults, truncate long lines, and consider returning counts or summaries before raw lines.
Make tools read-only by default. Log analysis rarely needs to mutate state. If a tool can write or restart services, gate it behind explicit confirmation on the client side.
Handle parse failures gracefully. Real logs are messy. Return structured error messages the model can reason about rather than raising exceptions that abort the session.
Use resources for static context, tools for dynamic queries. Expose a small fixed log file as a resource, but expose search over a streaming backend as a tool — the distinction helps the model choose the right access pattern.
Log the tool calls. Record every call_tool invocation and its arguments. This is invaluable for debugging agent behavior and auditing what the LLM actually did during an incident.

Conclusion

Building a log analysis agent with MCP gives you a clean separation between the LLM's reasoning layer and your observability stack. The server encapsulates where logs live and how to query them; the client handles conversation and tool orchestration; the model does what it's good at — interpreting patterns, correlating events, and explaining incidents in plain language. Because MCP is a standard, the same server you built here can power a CLI tool today and a Slack bot tomorrow without rewriting the integration layer. Start with the file-based server to validate your agent's behavior, then progressively swap in real backends like Loki, Elasticsearch, or CloudWatch as your needs grow. The investment in a protocol-based architecture pays off the moment your log sources change and your agent doesn't have to.