Build and Deploy a Custom MCP Server on SingleStore Cloud Functions

1%%sql2DROP DATABASE IF EXISTS mcp_demo;3CREATE DATABASE mcp_demo;4
5USE mcp_demo;6
7CREATE TABLE products (8    id INT PRIMARY KEY,9    name VARCHAR(255),10    category VARCHAR(100),11    price DECIMAL(10,2),12    stock INT,13    description TEXT14);15
16INSERT INTO products VALUES17(1, 'SingleStore T-Shirt', 'Apparel', 29.99, 150, 'Comfortable cotton tee with the SingleStore logo'),18(2, 'Developer Hoodie', 'Apparel', 59.99, 75, 'Warm hoodie for late-night coding sessions'),19(3, 'Mechanical Keyboard', 'Electronics', 149.99, 30, 'Cherry MX Blue switches, RGB backlight'),20(4, 'Ergonomic Mouse', 'Electronics', 79.99, 50, 'Vertical design reduces wrist strain'),21(5, 'Standing Desk', 'Furniture', 499.99, 20, 'Electric height-adjustable desk'),22(6, 'Monitor Light Bar', 'Electronics', 39.99, 100, 'Reduce eye strain with screen-mounted lighting'),23(7, 'Cable Management Kit', 'Accessories', 19.99, 200, 'Keep your desk clean and organized'),24(8, 'Laptop Stand', 'Accessories', 34.99, 80, 'Aluminum, adjustable height and angle'),25(9, 'Noise Cancelling Headphones', 'Electronics', 299.99, 25, 'ANC with 30-hour battery life'),26(10, 'Webcam HD Pro', 'Electronics', 89.99, 60, '1080p with auto-focus and built-in mic');27
28CREATE TABLE orders (29    id INT PRIMARY KEY,30    product_id INT,31    customer_name VARCHAR(255),32    quantity INT,33    order_date DATE,34    status VARCHAR(50)35);36
37INSERT INTO orders VALUES38(1, 1, 'Alice', 2, '2025-04-01', 'shipped'),39(2, 3, 'Bob', 1, '2025-04-02', 'delivered'),40(3, 5, 'Charlie', 1, '2025-04-03', 'processing'),41(4, 2, 'Diana', 3, '2025-04-04', 'shipped'),42(5, 9, 'Eve', 1, '2025-04-05', 'delivered'),43(6, 4, 'Frank', 2, '2025-04-06', 'processing'),44(7, 6, 'Grace', 4, '2025-04-07', 'shipped'),45(8, 10, 'Hank', 1, '2025-04-08', 'delivered'),46(9, 7, 'Ivy', 5, '2025-04-09', 'shipped'),47(10, 8, 'Jack', 1, '2025-04-10', 'processing');

1%pip install --quiet \2    "anyio>=4.10" \3    "starlette<0.47.0,>=0.40.0" \4    "fastmcp"5
6# Clear any stale imports from the runtime's default versions of these7# libraries so the freshly installed ones are picked up without a kernel8# restart. `anyio` in particular is loaded early by `singlestoredb.apps`,9# so its `abc` submodule is cached before pip upgrades the files on disk.10import sys11for _mod in list(sys.modules):12    if any(13        _mod == p or _mod.startswith(p + ".")14        for p in ("mcp", "anyio", "starlette", "fastmcp", "sniffio", "httpx")15    ):16        del sys.modules[_mod]17
18
19# Verify the versions we actually loaded.20from importlib.metadata import version as _pkg_version21import fastmcp, mcp, anyio, starlette  # force import; missing pkg fails loudly22for _pkg in ("fastmcp", "mcp", "anyio", "starlette"):23    print(f"{_pkg:<9} == {_pkg_version(_pkg)}")24print("Dependencies ready.")

1import singlestoredb as s22import logging3import json4
5logging.basicConfig(level=logging.INFO)6
7def run_query(query: str, params: dict | None = None) -> list[dict]:8    """Execute a SQL query and return results as a list of dicts."""9    with s2.connect(connection_url, database="mcp_demo") as conn:10        with conn.cursor() as cur:11            cur.execute(query, params or {})12            if cur.description:13                columns = [desc[0] for desc in cur.description]14                rows = cur.fetchall()15                return [dict(zip(columns, row)) for row in rows]16            return []17
18logging.info("Database connection ready.")

1from fastmcp import FastMCP2
3mcp = FastMCP(name="SingleStore Product Catalog")4
5
6@mcp.tool()7def health() -> str:8    """9    Liveness check for the SingleStore MCP server. Returns 'ok' if the10    server is reachable and SQL connections are working.11    """12    try:13        run_query("SELECT 1 AS ok")14        return "ok"15    except Exception as e:16        return f"error: {e}"17
18
19@mcp.tool()20def search_products(query: str, category: str = "") -> str:21    """22    Search the product catalog by name or description.23    Optionally filter by category.24
25    Args:26        query: Search term to match against product name or description27        category: Optional category filter (e.g., 'Electronics', 'Apparel')28    """29    sql = """30        SELECT id, name, category, price, stock, description31        FROM products32        WHERE (name LIKE %(query)s OR description LIKE %(query)s)33    """34    params = {"query": f"%{query}%"}35    if category:36        sql += " AND category = %(category)s"37        params["category"] = category38
39    results = run_query(sql, params)40    if not results:41        return f"No products found matching '{query}'."42    return json.dumps(results, indent=2, default=str)43
44
45@mcp.tool()46def get_product_details(product_id: int) -> str:47    """48    Get full details for a specific product by its ID.49
50    Args:51        product_id: The numeric ID of the product52    """53    results = run_query(54        "SELECT * FROM products WHERE id = %(product_id)s",55        {"product_id": product_id},56    )57    if not results:58        return f"No product found with ID {product_id}."59    return json.dumps(results[0], indent=2, default=str)60
61
62@mcp.tool()63def check_inventory(category: str = "") -> str:64    """65    Check current inventory levels. Optionally filter by category.66    Returns products sorted by stock level (lowest first).67
68    Args:69        category: Optional category filter70    """71    sql = "SELECT name, category, stock, price FROM products"72    params = {}73    if category:74        sql += " WHERE category = %(category)s"75        params["category"] = category76    sql += " ORDER BY stock ASC"77    return json.dumps(run_query(sql, params), indent=2, default=str)78
79
80@mcp.tool()81def get_recent_orders(limit: int = 10) -> str:82    """83    Get the most recent orders with product details.84
85    Args:86        limit: Number of orders to return (default 10)87    """88    sql = """89        SELECT o.id AS order_id, p.name AS product,90               o.customer_name, o.quantity, o.order_date, o.status91        FROM orders o92        JOIN products p ON o.product_id = p.id93        ORDER BY o.order_date DESC94        LIMIT %(limit)s95    """96    return json.dumps(run_query(sql, {"limit": limit}), indent=2, default=str)97
98
99@mcp.tool()100def get_sales_summary() -> str:101    """102    Get a summary of sales by category, including103    total revenue and units sold.104    """105    sql = """106        SELECT p.category,107               COUNT(o.id) AS total_orders,108               SUM(o.quantity) AS total_units,109               ROUND(SUM(o.quantity * p.price), 2) AS total_revenue110        FROM orders o111        JOIN products p ON o.product_id = p.id112        GROUP BY p.category113        ORDER BY total_revenue DESC114    """115    return json.dumps(run_query(sql), indent=2, default=str)116
117
118logging.info("MCP server created with tools registered.")

1from fastapi import FastAPI, Request2from fastapi.responses import Response3
4# Get fastmcp's ASGI handler — this is the real MCP protocol implementation5# that handles initialize, tools/list, tools/call, SSE streaming, etc.6mcp_asgi_app = mcp.http_app(path="/mcp")7_fastmcp_mcp_route = next(8    r for r in mcp_asgi_app.routes if getattr(r, "path", None) == "/mcp"9)10_mcp_asgi_handler = _fastmcp_mcp_route.app11
12app = FastAPI(13    title="SingleStore MCP Server",14    description="A custom MCP server deployed as a SingleStore Cloud Function",15    version="1.0.0",16    lifespan=mcp_asgi_app.lifespan,17)18
19
20class ASGIProxyResponse(Response):21    """A FastAPI/Starlette `Response` whose `__call__` delegates the entire22    ASGI interaction (headers, body, SSE streaming) to another ASGI app.23
24    Why: the SingleStore Cloud Function gateway only proxies requests to25    paths it sees in `openapi.json`. `fastmcp` registers `/mcp` as a plain26    Starlette `Route`, which FastAPI's schema generator never picks up —27    so the gateway 307-loops requests (`/mcp` ↔ `/mcp/`, prefix stripped).28    Adding an `@app.post("/mcp")` stub fixes the gateway's view but then29    the stub intercepts the request instead of `fastmcp`. Returning this30    proxy response from the stub hands the ASGI scope/receive/send off to31    `fastmcp` untouched, so SSE streaming and all MCP protocol semantics32    work end-to-end.33
34    We buffer the request body at handler time (via `request.body()`) and35    replay it on fastmcp's first `receive()` call. Subsequent `receive()`36    calls fall through to the real client connection so SSE keepalive and37    disconnect signals work normally.38    """39
40    def __init__(self, target, scope, body: bytes, real_receive):41        super().__init__(content=b"", status_code=200)42        self._target = target43        self._scope = scope44        self._body = body45        self._real_receive = real_receive46
47    async def __call__(self, scope, receive, send):48        replayed = False49
50        async def _receive():51            nonlocal replayed52            if not replayed:53                replayed = True54                return {"type": "http.request", "body": self._body, "more_body": False}55            return await self._real_receive()56
57        await self._target(self._scope, _receive, send)58
59
60@app.post("/mcp")61@app.get("/mcp", include_in_schema=False)62@app.delete("/mcp", include_in_schema=False)63async def mcp_endpoint(request: Request):64    """MCP Streamable HTTP endpoint — delegates to fastmcp's ASGI handler."""65    body = await request.body()66    return ASGIProxyResponse(67        _mcp_asgi_handler, request.scope, body, request.receive68    )

1import singlestoredb.apps as apps2
3connection_info = await apps.run_function_app(app)