How to Build an MCP Server for Your SIEM: Safe Log Access for AI Agents

The fastest way to turn a promising SOC copilot into an incident is to hand it a database connection string and call it “tool access.”

Why this matters now

Every SOC vendor is racing to bolt an LLM onto their SIEM. The easy version — give the model a run_sql tool and point it at your log warehouse — works in a demo and fails in production. It has no concept of tenant boundaries, no query cost limits, and no audit trail of what it actually read. In a multi-tenant platform, that is not a rough edge. It is a data-isolation bug waiting for the right prompt.

The Model Context Protocol (MCP) gives you a better pattern. Instead of exposing your database, you expose a small set of typed, purpose-built tools — search_cloudtrail_events, get_entity_timeline — and let the LLM call those. The server enforces scope, limits, and logging; the model never sees a connection string.

This guide walks through building an MCP server for a ClickHouse-backed SIEM, using the same principles a real investigation agent needs: read-only access, hard tenant isolation, and every call auditable after the fact.

The short version

What MCP actually gives you

MCP is a standard way for an LLM client (Claude Code, an agent runtime, a chat UI) to discover and call tools exposed by a separate server process, over stdio or HTTP. The client sends a tool name and arguments; the server validates them, does the work, and returns structured data. The protocol itself doesn’t grant safety — a run_sql(query: str) tool over MCP is exactly as dangerous as one wired up by hand. The safety comes from how you design the tool surface.

For a SIEM, that means resisting the urge to expose your query engine directly and instead exposing the handful of operations an analyst or investigation agent actually performs:

• Search events by entity (user, IP, host) within a time window
• Pull the full timeline for a specific entity around an alert
• Look up enrichment (GeoIP, threat intel match) for an IP or domain
• Fetch a specific event by ID for detail

Each of those is a bounded, auditable operation. None of them can be abused to enumerate other tenants’ data or run an unbounded scan.

Designing the tool surface

Start from the investigation workflows your analysts already run, not from your schema. If a human SOC analyst’s first move on a suspicious login alert is “show me this user’s activity for the last 24 hours,” that’s your tool — not a generic query_clickhouse.

# src/soc_agent/mcp/tools.py
from datetime import datetime, timedelta
from dataclasses import dataclass

@dataclass
class ToolContext:
    """Bound at server startup — never supplied by the model."""
    tenant_db: str          # e.g. "customer_acme"
    clickhouse_client: object
    max_rows: int = 500
    max_lookback_hours: int = 168  # 7 days


def search_events_by_entity(
    ctx: ToolContext,
    entity_type: str,       # "user" | "ip" | "host"
    entity_value: str,
    hours: int = 24,
    event_types: list[str] | None = None,
) -> dict:
    hours = min(hours, ctx.max_lookback_hours)
    since = datetime.utcnow() - timedelta(hours=hours)

    column = {"user": "actor", "ip": "source_ip", "host": "hostname"}.get(entity_type)
    if column is None:
        return {"error": f"unsupported entity_type: {entity_type}"}

    query = f"""
        SELECT event_time, type, event_name, actor, source_ip, summary
        FROM {ctx.tenant_db}.events
        WHERE {column} = %(entity_value)s
          AND event_time >= %(since)s
          {"AND type IN %(event_types)s" if event_types else ""}
        ORDER BY event_time DESC
        LIMIT %(limit)s
    """
    params = {"entity_value": entity_value, "since": since, "limit": ctx.max_rows}
    if event_types:
        params["event_types"] = tuple(event_types)

    rows = ctx.clickhouse_client.query(query, parameters=params)
    return {"row_count": len(rows), "events": rows, "truncated": len(rows) == ctx.max_rows}

Three things are doing the actual security work here, none of them visible to the model:

1. tenant_db is bound at process start, from the customer’s dedicated ClickHouse database — the same isolation the rest of the platform already relies on. The model passes an entity value; it never passes or sees a database name.
2. entity_type maps to an explicit allowlist of columns, not a raw field name from the caller. There’s no path from model input to an arbitrary SELECT column FROM table.
3. Every numeric input is clamped (min(hours, max_lookback_hours), LIMIT max_rows) before it reaches the query. An agent that asks for “everything since 2020” gets seven days and a truncated: true flag, not a full table scan.

Wiring tools into the MCP server

# src/soc_agent/mcp/server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
import mcp.types as types

from .tools import ToolContext, search_events_by_entity, get_entity_timeline, lookup_ip_enrichment

def build_server(ctx: ToolContext) -> Server:
    server = Server("xpernix-siem-investigation")

    @server.list_tools()
    async def list_tools() -> list[types.Tool]:
        return [
            types.Tool(
                name="search_events_by_entity",
                description="Search recent SIEM events for a user, IP, or host.",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "entity_type": {"type": "string", "enum": ["user", "ip", "host"]},
                        "entity_value": {"type": "string"},
                        "hours": {"type": "integer", "default": 24},
                        "event_types": {"type": "array", "items": {"type": "string"}},
                    },
                    "required": ["entity_type", "entity_value"],
                },
            ),
            # get_entity_timeline, lookup_ip_enrichment defined the same way
        ]

    @server.call_tool()
    async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
        audit_log.info("mcp_tool_call", tool=name, args=arguments, tenant=ctx.tenant_db)

        if name == "search_events_by_entity":
            result = search_events_by_entity(ctx, **arguments)
        else:
            result = {"error": f"unknown tool: {name}"}

        return [types.TextContent(type="text", text=json.dumps(result, default=str))]

    return server

The audit_log.info(...) call before dispatch is not optional. It’s the line that turns “the AI looked at something” into “the AI called search_events_by_entity for user [email protected] at 14:32 UTC, and here’s exactly what came back.” Ship this log line to the same ClickHouse cluster you already query, and an analyst reviewing an AI-assisted investigation can replay every tool call the agent made.

Guardrails that belong in the server, not the prompt

System prompts telling the model “only query your own tenant” or “don’t run expensive queries” are worth having, but they’re not a security boundary — they’re a suggestion to something that can be jailbroken or simply mistaken. Enforce the following in code:

That last row is what makes this different from a generic database chatbot: the tenant boundary and the query limits are structural, not conversational. An agent — or an attacker who’s found a way to influence its inputs — can ask for anything a tool’s schema allows and nothing more.

Where this fits with a triage or investigation agent

If you’re running an LLM-based triage or investigation pipeline (see our post on LLM-powered SOC alert triage), this MCP server is the piece that turns “the model summarized the alert” into “the model pulled the actual timeline, checked GeoIP on the source IP, and grounded its verdict in real data it’s allowed to see.” The investigation agent talks MCP to the server; the server talks parameterized SQL to ClickHouse. Neither side needs to trust the other with more than the protocol requires.

Final thought

The interesting design work in AI-assisted security tooling isn’t the model — it’s the boundary between the model and your data. MCP gives you a clean place to put that boundary: a small, typed, auditable tool surface instead of a database connection with a system prompt taped to it. Build the tools around what your analysts actually do, clamp every input, and log every call. The model gets exactly the access an investigation needs, and no more.

If you want help designing a safe tool surface for AI-assisted investigation on top of your own log pipeline, contact us.