MCP

MCP Server for Analytics: Letting Agents Query Your Data Warehouse

Pramod Dutta · May 27, 2026 · 14 min read

The 2 AM Slack message every data team dreads

"Hey, can you pull the signup numbers for the last 30 days broken out by plan tier?" You've seen this message a hundred times. You open a SQL editor, write the query, sanity-check the output, paste it into Slack, and move on with your day. Multiply that by every stakeholder who doesn't know SQL, and you've got a data team that spends half its week as a human query interface.

This is exactly the kind of repetitive, well-defined task that AI agents are good at — if you give them a safe way to reach your data. That's where the Model Context Protocol comes in. An MCP server for analytics doesn't just let an LLM "run some SQL." Done right, it gives an agent a controlled, auditable, schema-aware interface to your warehouse — Snowflake, BigQuery, Redshift, ClickHouse, or plain Postgres — so it can answer questions like a competent analyst instead of a chatbot that hallucinates numbers.

This article walks through why analytics is one of the best use cases for MCP, how to design the server so it doesn't turn into a security incident, and what a working implementation actually looks like in code. If you've been bolting "talk to my database" features onto chatbots with brittle prompt templates, this is the more durable pattern.

Why analytics is a natural fit for MCP

Model Context Protocol exists to solve a specific integration problem: every AI application that wants to talk to an external system used to need its own bespoke connector code, duplicated across every client. MCP standardizes that boundary. A server exposes tools (actions), resources (readable context), and optionally prompts (reusable templates), and any MCP-compatible client — Claude Desktop, Claude Code, a custom agent runtime — can use them without custom glue code.

Analytics is arguably the cleanest use case for this pattern, for a few reasons:

The domain is naturally tool-shaped. "Run this query," "list these tables," "describe this schema" map directly onto discrete, well-scoped functions — not vague open-ended actions.
Read-heavy by default. Most analytics questions are SELECT statements, not writes. That means the blast radius of a mistake is much smaller than, say, an agent with write access to a CRM.
Schema discovery is a real problem an LLM can help with. Large warehouses have hundreds of tables and inconsistent naming. An agent that can introspect the schema before writing a query produces far better SQL than one guessing from a prompt-injected schema dump.
The output is verifiable. Numbers from a warehouse can be cross-checked against a dashboard. This makes analytics a forgiving place to experiment with agentic workflows, because errors are usually caught quickly rather than silently trusted.

Contrast this with something like an MCP server that lets an agent send emails or modify production records — the cost of a wrong action is high and hard to reverse. A read-only analytics query that returns wrong numbers is annoying, not catastrophic, especially if you keep humans in the loop for anything that drives a real decision.

There's also a practical adoption argument. Data teams are usually stretched thin, and the bulk of their inbound requests are not novel analysis — they're variations on "how many X did we do last month" or "show me Y broken down by Z." Those questions have a shape. An agent with a well-designed analytics MCP server can absorb a large share of that repetitive volume, leaving analysts free for the harder, ambiguous work that actually needs a human's judgment: deciding which metric definition matters for a specific decision, spotting a data quality issue in the underlying pipeline, or pushing back on a question that's being asked because it's easy to ask rather than because it's the right thing to measure.

What the server actually needs to expose

A minimal but genuinely useful analytics MCP server needs three categories of tools.

Schema introspection tools. Before an agent can write a good query, it needs to know what tables and columns exist. Don't dump your entire information_schema into the model's context on every call — expose it as callable tools instead, so the agent only fetches what it needs.

list_tables — returns table names, optionally filtered by schema or a search term
describe_table — returns columns, types, and ideally short human-written descriptions
list_saved_queries (optional) — surfaces a library of vetted queries or views for common questions, which is often more reliable than freeform generation

Query execution tools. This is the core capability, and where the design work actually happens.

run_query — accepts a SQL string (or better, a structured query object) and returns rows, with the server enforcing timeouts, row limits, and query validation
explain_query — returns the query plan without executing, useful for catching runaway queries before they run

Metadata and governance tools. These make the server safe enough to actually deploy.

get_query_cost_estimate — for warehouses like BigQuery or Snowflake that expose cost/bytes-scanned estimates before execution
list_recent_queries — an audit tool so both agents and humans can see what's been run

Notice what's missing: there's no write_table, delete_rows, or run_arbitrary_ddl tool. That's not an oversight — it's the single most important design decision in this whole architecture.

Design principle one: the database user is your real security boundary

The most common mistake teams make when wiring an LLM up to a warehouse is treating the LLM's judgment as the security control. It isn't, and it never should be. Prompt instructions like "only run SELECT queries, never DROP or DELETE" are a start, but they're a suggestion to a probabilistic system, not an enforcement mechanism. An agent that gets a cleverly worded or accidentally malformed instruction, or that's manipulated via a prompt injection buried in a data value, can still attempt a write if nothing stops it downstream.

The fix is boring and well-understood: use database-level permissions.

-- Create a dedicated, read-only role for the MCP server
CREATE ROLE mcp_analytics_reader;

GRANT USAGE ON SCHEMA analytics TO mcp_analytics_reader;
GRANT SELECT ON ALL TABLES IN SCHEMA analytics TO mcp_analytics_reader;

-- Explicitly no INSERT, UPDATE, DELETE, DROP, ALTER grants

-- Create the service account the MCP server connects as
CREATE USER mcp_server_svc WITH PASSWORD '...';
GRANT mcp_analytics_reader TO mcp_server_svc;

-- Belt-and-suspenders: cap resource usage at the role level
ALTER ROLE mcp_analytics_reader SET statement_timeout = '30s';

With this in place, even if the LLM somehow generates DELETE FROM users, the database itself rejects it before any damage occurs. The MCP server's job is to be a well-behaved client of a properly restricted account — not to be the last line of defense. This is the same principle you'd apply to any third-party integration; an LLM-driven agent is just another untrusted caller from the database's point of view, and should be treated with the same skepticism as a public API endpoint.

Layer additional protections in the server itself, but treat them as defense in depth, not the primary control:

Reject any SQL containing INSERT, UPDATE, DELETE, DROP, ALTER, GRANT, TRUNCATE as a keyword-level pre-filter (cheap, catches obvious cases)
Enforce a hard row limit (LIMIT 1000 appended if missing) so a mistaken cross join doesn't return ten million rows into an LLM's context window
Enforce a query timeout at the connection-pool level, independent of the database role setting
Restrict which schemas are queryable at all — a marketing analytics agent has no business seeing a billing_internal schema even in read-only mode

Building the server: a working example

Let's make this concrete. Here's a Python MCP server using the official mcp SDK, connecting to a Postgres-compatible warehouse. This is intentionally simplified but structurally complete — the pieces you'd harden for production are called out afterward.

import asyncpg
import re
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

app = Server("analytics-warehouse")

FORBIDDEN_KEYWORDS = re.compile(
    r"\b(INSERT|UPDATE|DELETE|DROP|ALTER|GRANT|TRUNCATE|CREATE)\b",
    re.IGNORECASE,
)
ALLOWED_SCHEMAS = {"analytics", "marketing_public"}
MAX_ROWS = 1000
QUERY_TIMEOUT_SECONDS = 15

pool: asyncpg.Pool | None = None


async def get_pool() -> asyncpg.Pool:
    global pool
    if pool is None:
        pool = await asyncpg.create_pool(
            dsn="postgresql://mcp_server_svc:***@warehouse-host/prod",
            min_size=1,
            max_size=5,
        )
    return pool


@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="list_tables",
            description="List tables available in the analytics warehouse, optionally filtered by schema.",
            inputSchema={
                "type": "object",
                "properties": {
                    "schema": {"type": "string", "description": "Schema name to filter by"},
                },
            },
        ),
        Tool(
            name="describe_table",
            description="Get column names, types, and descriptions for a table.",
            inputSchema={
                "type": "object",
                "properties": {
                    "table_name": {"type": "string"},
                    "schema": {"type": "string"},
                },
                "required": ["table_name"],
            },
        ),
        Tool(
            name="run_query",
            description=(
                "Run a read-only SQL SELECT query against the analytics warehouse. "
                "Results are capped at 1000 rows and queries time out after 15 seconds."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "A single SELECT statement"},
                },
                "required": ["sql"],
            },
        ),
    ]


def validate_query(sql: str) -> str | None:
    """Return an error string if the query is disallowed, else None."""
    stripped = sql.strip().rstrip(";")
    if not stripped.lower().startswith("select"):
        return "Only SELECT statements are permitted."
    if FORBIDDEN_KEYWORDS.search(stripped):
        return "Query contains a disallowed keyword (writes/DDL are not permitted)."
    if ";" in stripped:
        return "Multiple statements are not permitted."
    return None


def enforce_row_limit(sql: str) -> str:
    if re.search(r"\blimit\s+\d+", sql, re.IGNORECASE):
        return sql
    return f"{sql.rstrip(';')} LIMIT {MAX_ROWS}"


@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    db = await get_pool()

    if name == "list_tables":
        schema_filter = arguments.get("schema")
        query = """
            SELECT table_schema, table_name
            FROM information_schema.tables
            WHERE table_schema = ANY($1)
            ORDER BY table_schema, table_name
        """
        schemas = [schema_filter] if schema_filter else list(ALLOWED_SCHEMAS)
        rows = await db.fetch(query, schemas)
        return [TextContent(type="text", text=str([dict(r) for r in rows]))]

    if name == "describe_table":
        table = arguments["table_name"]
        schema = arguments.get("schema", "analytics")
        if schema not in ALLOWED_SCHEMAS:
            return [TextContent(type="text", text=f"Schema '{schema}' is not accessible.")]
        query = """
            SELECT column_name, data_type, is_nullable
            FROM information_schema.columns
            WHERE table_schema = $1 AND table_name = $2
            ORDER BY ordinal_position
        """
        rows = await db.fetch(query, schema, table)
        return [TextContent(type="text", text=str([dict(r) for r in rows]))]

    if name == "run_query":
        sql = arguments["sql"]
        error = validate_query(sql)
        if error:
            return [TextContent(type="text", text=f"Query rejected: {error}")]

        safe_sql = enforce_row_limit(sql)
        try:
            async with db.acquire() as conn:
                async with conn.transaction():
                    await conn.execute(f"SET LOCAL statement_timeout = '{QUERY_TIMEOUT_SECONDS}s'")
                    rows = await conn.fetch(safe_sql)
            return [TextContent(type="text", text=str([dict(r) for r in rows]))]
        except Exception as exc:
            return [TextContent(type="text", text=f"Query failed: {exc}")]

    raise ValueError(f"Unknown tool: {name}")


async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

A few things worth noticing about this implementation. The row limit and timeout are enforced in application code as a second layer, on top of the database role restrictions from earlier — neither one alone is sufficient. The schema allowlist (ALLOWED_SCHEMAS) means even a perfectly well-behaved SELECT query against a schema you didn't intend to expose gets rejected before it reaches the database. And the keyword filter, while easy to bypass with enough SQL cleverness (comments, encoding tricks, CTEs that smuggle in writes via functions), is there to catch the 95% case cheaply — it's not pretending to be a complete defense, because the database-level read-only role behind it is.

Handling the schema problem: context without overload

The hardest part of making this actually useful isn't the plumbing — it's making sure the agent writes correct SQL against your specific schema. Warehouses accumulate cruft: tables named users_v2, columns called stat that actually mean status, timestamps stored as both created_at and created_ts in different tables from different eras.

Three things help significantly here:

Column and table descriptions matter more than you'd think. If your warehouse supports comments (COMMENT ON COLUMN analytics.orders.stat IS 'order status: pending, paid, refunded, cancelled'), populate them, and have describe_table surface them. This is a one-time investment that pays off every time an agent queries that table afterward.

Prefer curated views over raw tables where you can. Instead of pointing the agent at five raw tables it has to join correctly every time, build a mart_daily_signups view that already handles the join logic and grain, and mention it in the tool description. This is the same principle behind a data team building a semantic layer — you're doing the hard modeling work once instead of asking an LLM to rediscover it in every conversation.

Use MCP resources for static reference material. Alongside tools, MCP supports resources — read-only content a client can attach to context. A short markdown doc explaining your warehouse's grain conventions, common gotchas ("revenue in the orders table is gross, not net — use net_revenue for reporting"), and naming quirks can be exposed as a resource so the agent reads it once per session rather than you re-explaining it in every system prompt.

from mcp.types import Resource

@app.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="warehouse://docs/conventions",
            name="Warehouse conventions and gotchas",
            mimeType="text/markdown",
        ),
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "warehouse://docs/conventions":
        return (
            "# Warehouse Conventions\n\n"
            "- `net_revenue` excludes refunds and taxes; `revenue` does not.\n"
            "- `users.created_at` is UTC; `events.ts` is stored in America/New_York.\n"
            "- Use `mart_daily_signups` instead of joining `users` + `subscriptions` directly.\n"
        )
    raise ValueError(f"Unknown resource: {uri}")

Observability: knowing what your agent actually did

Once an agent can query your warehouse autonomously, you need visibility into what it ran, not just what it reported back. This matters for two reasons: catching bad queries before they become bad decisions, and catching abuse or prompt injection attempts before they become bigger problems.

At minimum, log every query the server executes, tagged with a session or user identifier, the tool that was called, the SQL that ran (post-validation, post-row-limit), the row count returned, and the execution time.

import logging
import time

logger = logging.getLogger("mcp.analytics.audit")

async def audited_execute(conn, sql: str, session_id: str):
    start = time.monotonic()
    try:
        rows = await conn.fetch(sql)
        logger.info(
            "query_executed",
            extra={
                "session_id": session_id,
                "sql": sql,
                "row_count": len(rows),
                "duration_ms": round((time.monotonic() - start) * 1000),
                "status": "success",
            },
        )
        return rows
    except Exception as exc:
        logger.warning(
            "query_failed",
            extra={"session_id": session_id, "sql": sql, "error": str(exc)},
        )
        raise

This log stream is worth piping into whatever your team already uses for observability, and it doubles as a debugging tool when an agent's answer doesn't match what you expected — you can go look at the exact query it ran instead of guessing.

Where this goes wrong in practice

A few failure modes show up repeatedly once teams start running these servers for real:

Silent wrong answers from ambiguous questions. "What's our churn rate?" means different things depending on whether you count logo churn, revenue churn, trailing 30 or 90 days, or annualized. An agent will confidently pick one definition and present it as *the* answer. Mitigate this by having curated views encode the definition (mart_monthly_logo_churn) rather than letting the agent freehand the calculation each time.
Cost blowups on usage-based warehouses. BigQuery and Snowflake charge by bytes scanned or compute time. An agent iterating on a query ("let me try without the filter to see the full data") can rack up cost quickly if there's no estimate-before-execute step. Expose a cost/bytes-estimate tool and make the agent's instructions require checking it before running against large tables.
Treating agent output as ground truth for high-stakes decisions. This pattern is genuinely useful for exploratory questions and quick pulls. It is not a replacement for a human reviewing the query behind a board-level revenue number. Keep that boundary explicit for your team.

Wrapping up

An MCP server for analytics is one of the more immediately practical applications of the protocol, because the underlying task — turning a natural-language question into a safe, scoped, verifiable query — plays to an LLM's strengths without requiring you to trust it with anything irreversible. The pattern that works is boring on purpose: a read-only database role as the real security boundary, application-level guardrails as a second layer, curated views to encode business logic instead of leaving it to the model, and an audit log so you can always see what actually ran.

Get those four things right, and you've turned your data warehouse from something only a few people can query into something your whole team — and their agents — can ask questions of directly, without anyone touching a DELETE statement by accident. If you want to go deeper on the protocol itself, how tool schemas are negotiated, and patterns for connecting agents to internal systems beyond just databases, that's exactly what we cover in Building & Integrating MCP Servers.