How to Build a Python MCP Server for Product Analytics
Your PM has a Slack message ready that starts with "can someone pull the numbers on...". This guide builds a read-only Python MCP server that answers two of those questions, feature usage and funnel counts, so an assistant can fetch them itself. MCP (Model Context Protocol) is the open standard that lets an AI client call functions you expose as tools. By the end you will have a server that runs over stdio, a client script that tests it, and a config that plugs it into Claude Desktop or Claude Code, with no database credentials anywhere near the assistant.
1. Set up the Python MCP server project
You need Python 3.11 or newer and uv, the Python package manager. The official build-a-server tutorial requires Python MCP SDK 2.0.0 or higher, and the cli extra adds the mcp command.
uv init analytics
cd analytics
uv add "mcp[cli]"
This creates the project and installs the SDK. You will add seed.py, server.py and test_client.py next to the generated files.
2. Seed a sample events table
In real life you would point the server at a read replica or a warehouse view. For now, a SQLite file with a few events is enough.
# seed.py
import sqlite3
conn = sqlite3.connect("analytics.db")
conn.executescript("""
DROP TABLE IF EXISTS events;
CREATE TABLE events (user_id TEXT, event TEXT, ts TEXT);
INSERT INTO events (user_id, event, ts) VALUES
('u1', 'signup', datetime('now', '-5 days')),
('u1', 'create_project', datetime('now', '-5 days')),
('u1', 'export_report', datetime('now', '-2 days')),
('u2', 'signup', datetime('now', '-4 days')),
('u2', 'create_project', datetime('now', '-4 days')),
('u3', 'signup', datetime('now', '-1 days'));
""")
conn.commit()
conn.close()
Run it once with uv run seed.py.
3. Write the read-only MCP tools
A tool is a plain Python function with the @mcp.tool() decorator. The SDK builds the tool definition from the type hints and the docstring, so write the docstring for the model, not for yourself.
# server.py
import logging
import sqlite3
from pathlib import Path
from typing import Annotated, Literal
from pydantic import BaseModel, Field
from mcp.server import MCPServer
from mcp.types import ToolAnnotations
logging.basicConfig(level=logging.INFO) # writes to stderr, safe for stdio
logger = logging.getLogger(__name__)
mcp = MCPServer("product-analytics")
DB_URI = Path(__file__).resolve().with_name("analytics.db").as_uri() + "?mode=ro"
Feature = Literal["create_project", "invite_teammate", "export_report"]
Funnel = Literal["onboarding", "collaboration"]
FUNNELS = {
"onboarding": ["signup", "create_project", "export_report"],
"collaboration": ["signup", "create_project", "invite_teammate"],
}
Days = Annotated[int, Field(ge=1, le=90, description="Look-back window in days.")]
READ_ONLY = ToolAnnotations(read_only_hint=True, open_world_hint=False)
def query(sql: str, params: tuple) -> list[sqlite3.Row]:
conn = sqlite3.connect(DB_URI, uri=True)
conn.row_factory = sqlite3.Row
try:
return conn.execute(sql, params).fetchall()
finally:
conn.close()
class FeatureUsage(BaseModel):
feature: str
days: int
unique_users: int
total_events: int
class FunnelStep(BaseModel):
step: str
users: int
class FunnelCounts(BaseModel):
funnel: str
days: int
steps: list[FunnelStep]
@mcp.tool(title="Feature usage", annotations=READ_ONLY)
def feature_usage(feature: Feature, days: Days = 30) -> FeatureUsage:
"""Unique users and total events for one product feature in the last N days."""
row = query(
"SELECT COUNT(DISTINCT user_id) AS users, COUNT(*) AS events "
"FROM events WHERE event = ? AND ts >= datetime('now', ?)",
(feature, f"-{days} days"),
)[0]
logger.info("feature_usage feature=%s days=%s", feature, days)
return FeatureUsage(
feature=feature, days=days,
unique_users=row["users"], total_events=row["events"],
)
@mcp.tool(title="Funnel counts", annotations=READ_ONLY)
def funnel_counts(funnel: Funnel, days: Days = 30) -> FunnelCounts:
"""Users who completed every step up to and including each step of a funnel
in the last N days. Event order is not checked."""
reached: set[str] | None = None
steps: list[FunnelStep] = []
for event in FUNNELS[funnel]:
rows = query(
"SELECT DISTINCT user_id FROM events "
"WHERE event = ? AND ts >= datetime('now', ?)",
(event, f"-{days} days"),
)
users = {r["user_id"] for r in rows}
reached = users if reached is None else reached & users
steps.append(FunnelStep(step=event, users=len(reached)))
logger.info("funnel_counts funnel=%s days=%s", funnel, days)
return FunnelCounts(funnel=funnel, days=days, steps=steps)
if __name__ == "__main__":
mcp.run(transport="stdio")
Three choices make this safe to hand to an assistant. The Literal types mean the model can only pick features and funnels you listed. The SQL is parameterized and the database opens with mode=ro, so a bad call cannot write. And the tools return aggregates, never user IDs.
The read_only_hint annotation only tells the client what to expect. The SDK tools docs say never to rely on a client honouring it, so the read-only connection (and a read-only database role in production) is the real guard.
4. Run the Python MCP server over stdio
uv run server.py
It prints nothing and does not return, which is correct. A host launches this process and talks to it over stdin and stdout, which is why stdout belongs to the protocol. Stop it with Ctrl-C.
5. Test the MCP tools with a client
The SDK's testing guide passes the server object straight to Client, so no subprocess or port is needed.
# test_client.py
import anyio
from mcp import Client
from server import mcp
async def main() -> None:
async with Client(mcp, raise_exceptions=True) as client:
tools = await client.list_tools()
print([tool.name for tool in tools.tools])
usage = await client.call_tool(
"feature_usage", {"feature": "export_report", "days": 30}
)
print(usage.structured_content)
funnel = await client.call_tool("funnel_counts", {"funnel": "onboarding"})
print(funnel.structured_content)
anyio.run(main)
Run uv run test_client.py. With the seed data, export_report shows one unique user and the onboarding funnel steps read 3, 2, 1. Keep raise_exceptions=True in tests only.
To test the real stdio path, give Client a StdioServerParameters instead of the server object:
from mcp import Client, StdioServerParameters
params = StdioServerParameters(
command="uv",
args=["--directory", "/absolute/path/to/analytics", "run", "server.py"],
)
async with Client(params) as client:
result = await client.list_tools()
6. Connect the MCP server to a client
For Claude Desktop, add the server under mcpServers in claude_desktop_config.json, then fully quit and reopen the app.
{
"mcpServers": {
"product-analytics": {
"command": "uv",
"args": ["--directory", "/absolute/path/to/analytics", "run", "server.py"]
}
}
}
For Claude Code, register it from the terminal:
claude mcp add product-analytics -- uv --directory /absolute/path/to/analytics run server.py
Then ask something like "how many people exported a report in the last 30 days?" and watch it call feature_usage.
Gotchas
- Keep stdout clean. A stray
print()corrupts the JSON-RPC stream on stdio. Uselogging, which writes to stderr. - Old tutorials use the old import. In SDK v2,
FastMCPfrommcp.server.fastmcpbecameMCPServerfrommcp.server, and the old path fails at import. - Hosts need absolute paths. They launch the server from their own working directory. If a host cannot find
uv, put the full path fromwhich uv(where uvon Windows) incommand. In the JSON file on Windows, use forward slashes. - The SDK client's stdio child does not inherit your environment. If you later read a connection string from an env var, pass it with
env=inStdioServerParameters. - The funnel ignores order. It counts users who did every step in the window, which is why the docstring says so. The model repeats whatever the docstring promises.
- Small segments can identify people. Add a minimum count before you add filters like country or plan.
A typed, read-only, tested MCP server gives product questions a safe path to live data, which is how you let an assistant into production analytics without handing it the keys.