MCP Ski Resort

A Python client library for interacting with Snowflake Cortex Agents, Cortex Search, and Cortex Analyst — featuring Alpine Mountain Ski Resort analytics and a Managed MCP Server.

Install

pip install -e ".[dev]"

# For async support (httpx)
pip install -e ".[async]"

Architecture

The notebooks are the source of truth for both the library code and documentation.

Snowflake-MCP/
├── 00_core.ipynb            # Library: JWT auth, SSE streaming, normalization, threads
├── 01_async.ipynb           # Library: Async transport (httpx)
├── 02_mcp_client.ipynb      # Library: MCP JSON-RPC client & display helpers
├── 03_agent_streaming.ipynb  # Demo: Cortex Agent SSE streaming
├── 04_see_pattern.ipynb     # Demo: Search-Enrich-Execute pattern
├── 05_mcp_client_demo.ipynb # Demo: Full MCP client walkthrough
├── index.ipynb              # This file → README.md
├── mcp_ski_resort/          # Python package (auto-generated by nbdev-export)
│   ├── core.py
│   ├── astream.py
│   └── mcp_client.py
├── sql/                     # DDL scripts for Snowflake setup
└── tests/                   # Integration test scripts

Notebooks 00–02 export the mcp_ski_resort library via #| export directives.
Notebooks 03–04 are demo notebooks that showcase the library API.

Library Modules

Module	Notebook	Highlights
`mcp_ski_resort.core`	`00_core`	JWT auth, `SnowflakeSession`, `stream_agent_sse`, `normalize_event`, `run_agent`, `AgentChat`, `create_thread`, thread-mode conversations
`mcp_ski_resort.astream`	`01_async`	Async equivalents via `httpx`: `async_run_agent`, `AsyncAgentChat`, `async_stream_agent_sse`. Install with `pip install mcp-ski-resort[async]`
`mcp_ski_resort.mcp_client`	`02_mcp_client`	MCP JSON-RPC client (`MCPClient`, `mcp_call`, `display_mcp_result`)

MCP Server Tools

The Alpine Mountain MCP Server (SKI_RESORT_MCP) exposes 10 tools across 5 Snowflake tool types:

Tool Name	Type	Description
`daily-summary-analyst`	CORTEX_ANALYST_TEXT_TO_SQL	Natural-language queries over daily operations
`revenue-analyst`	CORTEX_ANALYST_TEXT_TO_SQL	Revenue and financial analytics
`operations-analyst`	CORTEX_ANALYST_TEXT_TO_SQL	Operational metrics and staffing
`resort-executive-agent`	CORTEX_AGENT_RUN	Executive-level analysis with multi-step reasoning
`ski-ops-assistant`	CORTEX_AGENT_RUN	Operational assistant for ski patrol and lifts
`feedback-search`	CORTEX_SEARCH_SERVICE_QUERY	Semantic search over 3,937 guest feedback records
`incident-search`	CORTEX_SEARCH_SERVICE_QUERY	Semantic search over 2,834 safety incidents
`execute-sql`	SYSTEM_EXECUTE_SQL	Run ad-hoc SQL against AM_SKI_RESORT
`resort-kpi-summary`	GENERIC	KPI dashboard UDF (season parameter)
`weather-impact-report`	GENERIC	Weather impact analysis UDF (date range)
`weather-impact-report`	GENERIC	Weather impact analysis UDF

Quick Start

Sync — Agent streaming

from mcp_ski_resort.core import run_agent, AgentChat

# One-shot call
result = run_agent("RESORT_EXECUTIVE", "How did we perform this season?")
print(result.answer)

# Multi-turn conversation (local history)
chat = AgentChat("RESORT_EXECUTIVE")
chat.ask("What are our top KPIs?")
chat.ask("Compare with last season")  # history sent automatically

# Multi-turn with server-side threads
from mcp_ski_resort.core import create_thread
tid = create_thread()
chat = AgentChat("RESORT_EXECUTIVE", thread_id=tid)
chat.ask("What are our top KPIs?")  # server owns continuity

Async

from mcp_ski_resort.astream import async_run_agent, AsyncAgentChat

result = await async_run_agent("RESORT_EXECUTIVE", "Season summary")

MCP client

from mcp_ski_resort.mcp_client import mcp_tools_list, mcp_call, display_mcp_result

tools = mcp_tools_list()
resp = mcp_call("resort-executive-agent", {"message": "Season summary"})
display_mcp_result(resp, label="Executive Summary")

Prerequisites

See PREREQUISITES.md for full setup instructions including:

Snowflake account with Cortex Agents, Search, and Analyst
RSA key-pair authentication configured
SQL setup scripts in sql/ for creating search services, UDFs, and the MCP server