🔌 MCP Server Builder Starter — Ship Your First Production MCP Server

MCP Server Blueprint: Internal Snowflake Analytics MCP

An MCP server blueprint produces the complete production-grade design — primitives (tools/resources/prompts), transport, auth, schemas, rate limiting, and deployment — based on Anthropic's MCP spec + analysis of successful community servers (GitHub, Slack, Postgres, Notion MCP). Your use case is a classic enterprise remote MCP: internal users, authenticated, wrapping a controlled data source (Snowflake), with audit + compliance requirements. This blueprint produces the Python HTTP MCP server with 6 tools, 3 resources, 2 prompts, Okta OIDC auth, per-user rate limiting, full audit logging, and Claude Desktop integration — ready to ship in ~2 weeks of implementation.

Is MCP The Right Pattern?

MCP is appropriate here. Honest check:

For:

• Structured Claude access to internal data ✓
• Multi-user with centralized governance ✓
• Wraps existing system (Snowflake) — MCP's intended use case ✓
• Users already use Claude Desktop ✓

Against (arguments for alternatives):

• Could be a simple slash-command in a Slack bot (but loses Claude conversation context)
• Could be a Python library each analyst imports (but requires Python skills + less natural)
• Could be a web UI (but not integrated with Claude workflow)

Verdict: MCP is right. It gives analysts + PMs + execs conversational Claude access to data without SQL expertise, with central governance via the MCP server's auth layer.

Language + Transport Decision

Language: Python (per your preference + team skill)

• Use the mcp-python-sdk (Anthropic's official SDK)
• FastAPI for HTTP transport layer if using HTTP
• asyncio for concurrent query handling

Transport: HTTP (streamable)

• NOT stdio (you need remote hosted, multi-user)
• HTTP with SSE for streaming longer query results
• MCP 2025-06 spec supports this well

Architecture:

Claude Desktop (analyst's machine)
      ↓ HTTPS over company VPN or Cloudflare Zero Trust
MCP Server (AWS inside VPC)
      ↓ SQL (service account)
Snowflake Data Warehouse

Primitives Design: Tools, Resources, Prompts

TOOLS (actions):

• run_query — execute a SQL query with safety limits
• explain_query — Claude writes SQL, this explains what it would do without running
• get_query_cost — estimate cost before running expensive queries
• save_query_as_view — for analysts building reusable queries (with approval workflow)
• describe_table — get detailed schema + sample rows for a specific table
• list_recent_queries — show user's recent query history

RESOURCES (read-only data):

• snowflake://schemas — list of accessible schemas
• snowflake://tables/{schema} — list of tables in schema
• snowflake://views/prebuilt — catalog of pre-built views with descriptions

PROMPTS (reusable templates):

• weekly_metrics_report — pre-built prompt for weekly KPI report generation
• user_cohort_analysis — pre-built prompt for cohort analysis

Why this split:

• Tools = actions Claude executes (running queries has side effects: cost, audit log, rate limit consumption)
• Resources = reference data Claude can browse (schemas, table lists)
• Prompts = canned analytical templates users + Claude can reuse

Tool Definitions

Tool 1: `run_query`

@server.tool()
async def run_query(
    query: str,
    max_rows: int = 1000,
    timeout_seconds: int = 30
) -> ToolResult:
    """
    Execute a SQL query against the Snowflake warehouse. Use when the user
    asks a data question that requires fetching actual data (not schema
    exploration). Results are limited to max_rows (default 1000) and time-
    out after timeout_seconds (default 30, max 120). Query must be SELECT
    only — modifications are blocked at the server level.
    
    Returns: list of rows (JSON) + metadata about the query execution.
    
    Use `explain_query` first if the user is unsure what the query does.
    Use `get_query_cost` first for exploratory queries against large tables.
    """

Schema:

{
  "type": "object",
  "properties": {
    "query": {
      "type": "string",
      "description": "SQL SELECT query. Max 4000 chars. Must be read-only (SELECT only)."
    },
    "max_rows": {
      "type": "integer",
      "description": "Maximum rows to return (1-10000, default 1000)",
      "minimum": 1,
      "maximum": 10000,
      "default": 1000
    },
    "timeout_seconds": {
      "type": "integer",
      "description": "Query timeout (5-120 seconds, default 30)",
      "minimum": 5,
      "maximum": 120,
      "default": 30
    }
  },
  "required": ["query"]
}

Safety checks (before execution):

• Query parser validates SELECT-only (block INSERT, UPDATE, DELETE, DROP, CREATE, ALTER)
• Reject queries touching restricted schemas (audit_log, auth tables)
• Enforce max_rows at query level (add LIMIT if not present)
• Log query + user to audit before execution

Tool 2: `explain_query`

Returns what the query WOULD do (parsed + semantic analysis) without running it.

Tool 3: `get_query_cost`

Uses Snowflake EXPLAIN to estimate credits consumed. Essential for large-table queries.

Tool 4: `save_query_as_view`

Lets analysts save a query as a view. REQUIRES approval workflow (posts to #data-eng Slack for review before actual view creation).

Tool 5: `describe_table`

Returns: column names, types, sample 5 rows, foreign keys, row count, last updated.

Tool 6: `list_recent_queries`

Returns user's last 20 queries with timestamps + status. Helps Claude understand analyst's current thread.

Resource Definitions

Resource: `snowflake://schemas`

@server.resource("snowflake://schemas")
async def list_schemas() -> ResourceContent:
    # Returns JSON: { schemas: [{name, table_count, description}, ...] }

Resource: `snowflake://tables/{schema}`

@server.resource("snowflake://tables/{schema}")
async def list_tables(schema: str) -> ResourceContent:
    # Returns JSON: { tables: [{name, row_count, description, last_updated}, ...] }

Resource: `snowflake://views/prebuilt`

Catalog of 15 pre-built analytical views with descriptions. The 'menu' of curated queries.

Prompt Definitions

Prompt: `weekly_metrics_report`

Pre-composed analytical prompt that:

• Queries weekly active users, revenue, new signups, churn
• Compares to previous week + 4-week average
• Identifies notable changes
• Produces markdown report ready to paste into Notion

Prompt: `user_cohort_analysis`

Structured cohort analysis prompt that analysts can invoke with specific date ranges.

Auth + Security Architecture

User auth: Okta OIDC

Flow:

1. User opens Claude Desktop, MCP server configured in settings

2. First tool call: server requires OIDC token (user has already authenticated via Okta in Claude Desktop's OAuth flow)

3. Server validates token signature + issuer + expiry via Okta JWKS

4. Server checks user is member of 'data-users' Okta group

5. If valid: proceed. If invalid: return isError with auth-failure message.

Server-to-Snowflake auth: Service account

• Dedicated Snowflake service account with role DATA_ANALYST_READ_ONLY
• Role has access to relevant schemas ONLY (no PII schemas containing raw user data)
• Service account credentials in AWS Secrets Manager
• Rotated quarterly

Data isolation:

• No user token ever reaches Snowflake (server uses its own service account)
• User queries are audit-logged with user_id from Okta token
• Queries cannot escape allowed schemas (enforced at SQL parsing)

Network isolation:

• MCP server in private AWS subnet
• Access only via company VPN or Cloudflare Zero Trust
• No public internet exposure
• Snowflake connection via PrivateLink (no public network path)

Error Response Contracts

All errors return structured:

{
  "isError": true,
  "content": [
    {
      "type": "text",
      "text": "Clear error message Claude can reason about."
    }
  ],
  "meta": {
    "error_code": "AUTH_EXPIRED",
    "retry_advice": "Reconnect via Okta."
  }
}

Error categories:

• AUTH_EXPIRED — user needs re-auth
• PERMISSION_DENIED — user not in allowed group
• RATE_LIMITED — user exceeded quota
• QUERY_INVALID — SQL parsing or safety violation
• QUERY_TIMEOUT — exceeded timeout
• QUERY_COST_EXCEEDED — preflight cost check blocked query
• UPSTREAM_ERROR — Snowflake unavailable or returned error

Claude can reason about these to suggest appropriate next actions ('let me break this query into smaller pieces' for timeout, etc.).

Rate Limiting + Quotas

Per-user quotas:

• 50 queries per hour (enforced via Redis counter keyed by user_id)
• 500 queries per day
• Max 5 concurrent queries per user

Cost quotas:

• 10 Snowflake credits per user per day
• Preflight check via get_query_cost if query is estimated >0.5 credit
• Queries estimated >2 credits require explicit user confirmation via tool meta

Global rate limit:

• 100 queries/minute total server-wide (protects Snowflake from DDoS from us)
• Circuit breaker: if Snowflake returns 5+ errors in 60s, pause all queries 5 min

Testing Plan

Local development testing:

1. Run server locally with `uvicorn server:app --reload`

2. Use `mcp inspector` (official Anthropic tool) to probe all tools/resources/prompts

3. Test each tool with valid + invalid inputs

4. Test each error category

Staging environment testing:

1. Deploy to staging AWS env (staging Snowflake copy)

2. Integrate with Claude Desktop config

3. Run through 15 real analyst workflows with team

4. Stress test: 35 concurrent simulated users

Security testing:

1. Auth bypass attempts (expired tokens, fake tokens, wrong group)

2. SQL injection attempts (despite SELECT-only — test parser robustness)

3. Schema escape attempts (try querying restricted schemas)

4. Rate limit bypass attempts

Pre-production checklist:

• All tool schemas validated via JSON Schema validator
• All resources return valid content types
• Error responses match contract for every error category
• Audit log entries created for every query
• Metrics exported to Datadog/Grafana
• Runbook documented for on-call

Deployment + Distribution

Server deployment:

• Docker image pushed to internal ECR
• Deployed to ECS Fargate in private VPC
• 2 tasks minimum, auto-scale to 5 based on CPU
• ALB in front with ACM cert
• CloudFlare Zero Trust for user access

Infrastructure:

• Secrets in AWS Secrets Manager (Snowflake service account, Okta config)
• Redis (ElastiCache) for rate limit counters
• CloudWatch + Datadog for monitoring
• Audit logs to S3 + Athena for queryable history

Distribution to users:

• NOT via npm/PyPI — internal only
• Users add entry to their Claude Desktop config
• IT pushes the config via MDM to user machines
• README in internal wiki with setup instructions

Claude Desktop/Code Integration

Claude Desktop config entry:

{
  "mcpServers": {
    "snowflake-analytics": {
      "url": "https://snowflake-mcp.internal.company.com",
      "auth": {
        "type": "oauth",
        "okta_issuer": "https://company.okta.com"
      }
    }
  }
}

Install instructions (internal wiki):

1. Open Claude Desktop settings

2. Navigate to 'Developer' → 'MCP Servers'

3. Add 'snowflake-analytics' with config above

4. Authenticate via Okta when prompted

5. Restart Claude Desktop

6. Test with: 'List available data schemas'

Troubleshooting (for wiki):

• Auth errors → re-run Okta auth
• Permission denied → request 'data-users' group access
• Rate limited → wait 1 hour or contact data-eng
• Query timeout → use get_query_cost before running

Key Takeaways

• 6 tools + 3 resources + 2 prompts. Python HTTP MCP server. Okta OIDC auth. Service-account access to Snowflake. Private deployment via Cloudflare Zero Trust.
• Primitive split: TOOLS = actions with side effects (queries cost money/time/audit). RESOURCES = read-only reference (schemas, tables). PROMPTS = reusable analytical templates.
• Auth is your responsibility: user → Okta OIDC → MCP server → (with service account) → Snowflake. User token never reaches Snowflake.
• Rate limiting + cost quotas are non-negotiable. 50 queries/hour/user, 10 Snowflake credits/day/user, 100 queries/min server-wide.
• Test with mcp inspector first, then Claude Desktop integration. Security test auth bypass + SQL injection + schema-escape attempts before production.