Build a ChatGPT App with a self-hosted MCP server (Python)

mkdir coveo-mcp-server && cd coveo-mcp-server
python -m venv .venv
source .venv/bin/activate

mcp[cli]>=1.26.0
httpx
python-dotenv

pip install -r requirements.txt

COVEO_API_KEY=<YOUR_API_KEY>
COVEO_ORGANIZATION_ID=<YOUR_ORG_ID>
COVEO_COMMERCE_TRACKING_ID=<YOUR_TRACKING_ID>
COVEO_COMMERCE_CLIENT_ID=<YOUR_CLIENT_ID>
COVEO_COMMERCE_VIEW_URL=https://www.example.com/

import os
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP 
from pathlib import Path
from commerce_search import make_commerce_search_request 

load_dotenv() 

API_KEY = os.environ["COVEO_API_KEY"]
ORG_ID = os.environ["COVEO_ORGANIZATION_ID"]
TRACKING_ID = os.environ["COVEO_COMMERCE_TRACKING_ID"]
CLIENT_ID = os.environ["COVEO_COMMERCE_CLIENT_ID"]
VIEW_URL = os.environ.get("COVEO_COMMERCE_VIEW_URL", "https://www.example.com/")
COMMERCE_ENDPOINT = (
    f"https://platform.cloud.coveo.com/rest/organizations/{ORG_ID}/commerce/v2/search"
) 

mcp = FastMCP("coveo-mcp-server")

COMMERCE_APP_URI = "ui://coveo/commerce-search"
COMMERCE_APP_HTML = Path("commerce-search.html").read_text() if Path("commerce-search.html").exists() else None 

TOOL_DESCRIPTION = (
    "Search products in Coveo Commerce and return normalized product results. "
    "When the shopper provides structured constraints such as budget, price ceiling, "
    "or product attributes, keep the natural-language query concise and pass those "
    "constraints through facetFilters. Use range filters for price or budget "
    "constraints (for example, field ec_price with max 700) and value filters for "
    "other attributes. The available facet fields depend on the commerce fields "
    "indexed in the Coveo source."
) 


@mcp.tool(
    description=TOOL_DESCRIPTION,
    metadata={"ui": {"resourceUri": COMMERCE_APP_URI}} if COMMERCE_APP_HTML else {}, 
)
async def commerce_search_coveo(
    query: str,
    numberOfResults: int = 5,
    page: int = 0,
    facetFilters: list[dict] | None = None, 
) -> dict:
    return await make_commerce_search_request( 
        query, numberOfResults, page, facetFilters,
        api_key=API_KEY,
        endpoint=COMMERCE_ENDPOINT,
        tracking_id=TRACKING_ID,
        client_id=CLIENT_ID,
        view_url=VIEW_URL,
    )


if COMMERCE_APP_HTML:

    @mcp.resource(
        uri=COMMERCE_APP_URI,
        name="Coveo Commerce Search App",
        mime_type="text/html;profile=mcp-app", 
    )
    async def commerce_search_app_resource() -> str:
        return COMMERCE_APP_HTML


if __name__ == "__main__":
    mcp.run(transport="streamable-http", stateless_http=True) 

import re
import httpx 

BUDGET_PATTERNS = [
    re.compile(
        r"(?:budget(?:\s+(?:is|of|around|about|under|below|up\s+to|max(?:imum)?))?\s*\$?)"
        r"(?P<amount>\d[\d,]*(?:\.\d+)?)\b",
        re.IGNORECASE,
    ),
    re.compile(
        r"(?:under|below|less\s+than|up\s+to|max(?:imum)?(?:\s+budget)?|price\s+ceiling)"
        r"\s*\$?(?P<amount>\d[\d,]*(?:\.\d+)?)\b",
        re.IGNORECASE,
    ),
    re.compile(r"\$\s*(?P<amount>\d[\d,]*(?:\.\d+)?)"),
] 


def extract_budget_max(query: str) -> float | None:
    for pattern in BUDGET_PATTERNS:
        m = pattern.search(query)
        if m:
            return float(m.group("amount").replace(",", ""))
    return None


def build_facets_payload(
    facet_filters: list[dict],
) -> list[dict]: 
    facets = []
    for f in facet_filters:
        if f.get("type") == "range":
            facets.append(
                {
                    "facetId": f["field"],
                    "field": f["field"],
                    "type": "numericalRange",
                    "values": [
                        {
                            "start": f.get("min", 0),
                            "end": f["max"],
                            "endInclusive": True,
                            "state": "selected",
                        }
                    ],
                }
            )
        elif f.get("type") == "value":
            facets.append(
                {
                    "facetId": f["field"],
                    "field": f["field"],
                    "type": "regular",
                    "values": [
                        {"value": v, "state": "selected"} for v in f.get("values", [])
                    ],
                }
            )
    return facets


async def make_commerce_search_request( 
    query: str,
    num_results: int = 5,
    page: int = 0,
    facet_filters: list[dict] | None = None,
    *,
    api_key: str,
    endpoint: str,
    tracking_id: str,
    client_id: str,
    view_url: str,
) -> dict:
    effective_filters = list(facet_filters or [])
    budget_max = extract_budget_max(query)

    has_explicit_price = any(
        f.get("type") == "range" and f.get("field") == "ec_price"
        for f in effective_filters
    )
    if budget_max is not None and not has_explicit_price:
        effective_filters.append(
            {"type": "range", "field": "ec_price", "min": 0, "max": budget_max}
        ) 

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }
    payload = {
        "query": f"<@- {query.strip()} -@>", 
        "perPage": num_results,
        "page": page,
        "trackingId": tracking_id, 
        "clientId": client_id,
        "country": "US",
        "currency": "USD",
        "language": "en",
        "context": {"view": {"url": view_url}},
    }

    if effective_filters:
        payload["facets"] = build_facets_payload(effective_filters)

    async with httpx.AsyncClient() as client:
        response = await client.post(
            endpoint,
            headers=headers,
            json=payload,
            timeout=30.0,
        )
        response.raise_for_status() 

    data = response.json()
    products = [
        {
            "productId": p.get("ec_product_id"),
            "name": p.get("ec_name"), 
            "brand": p.get("ec_brand"),
            "description": p.get("ec_description"),
            "imageUrl": p.get("ec_images"),
            "clickUri": p.get("clickUri"),
            "price": p.get("ec_price"),
            "currency": "USD",
        }
        for p in data.get("products", [])
    ]

    if budget_max is not None and products:
        products = [p for p in products if p["price"] is None or p["price"] <= budget_max] 

    result = {
        "query": query,
        "page": page,
        "perPage": num_results,
        "totalCount": data.get("totalCount", 0),
        "products": products,
    }
    if budget_max is not None:
        result["appliedConstraints"] = {"priceMax": budget_max}
    return result

python server.py

npm create vite@latest coveo-commerce-app -- --template react-ts
cd coveo-commerce-app
npm install @modelcontextprotocol/sdk @modelcontextprotocol/ext-apps vite-plugin-singlefile

import react from '@vitejs/plugin-react';
import {defineConfig} from 'vite';
import {viteSingleFile} from 'vite-plugin-singlefile'; 

export default defineConfig({
  plugins: [react(), viteSingleFile()],
});

import {useApp} from '@modelcontextprotocol/ext-apps/react'; 
import type {CallToolResult} from '@modelcontextprotocol/sdk/types.js';
import {useState} from 'react';

interface Product {
  productId: string | null;
  name: string | null;
  brand: string | null;
  description: string | null;
  imageUrl: string | null;
  clickUri: string | null;
  price: number | null;
  currency: string | null;
}

type Phase = 'waiting' | 'loading' | 'ready' | 'error';

function CommerceApp() {
  const [phase, setPhase] = useState<Phase>('waiting');
  const [products, setProducts] = useState<Product[]>([]);

  const {error} = useApp({
    appInfo: {name: 'coveo-commerce-app', version: '1.0.0'},
    capabilities: {},
    onAppCreated: (app) => {
      app.ontoolinput = () => setPhase('loading'); 

      app.ontoolresult = (result: CallToolResult) => {
        try {
          const text = result.content?.find((c) => c.type === 'text')?.text; 
          const data = text ? (JSON.parse(text) as {products?: Product[]}) : null;
          setProducts(data?.products ?? []);
          setPhase('ready');
        } catch {
          setPhase('error');
        }
      };
    },
  }); 

  if (error) return <p>Something went wrong: {error.message}</p>;
  if (phase === 'waiting') return <p>Waiting for results...</p>;
  if (phase === 'loading') return <p>Searching products...</p>;
  if (phase === 'error') return <p>Something went wrong.</p>;

  const formatPrice = (price: number | null, currency: string | null) => {
    if (price === null) return '';
    try {
      return new Intl.NumberFormat('en-US', {
        style: 'currency',
        currency: currency ?? 'USD',
      }).format(price); 
    } catch {
      return `$${price.toFixed(2)}`;
    }
  };

  return (
    <div style={{display: 'flex', flexDirection: 'column', gap: '1rem'}}>
      {products.map((product: Product) => (
        <article
          key={product.productId}
          style={{
            display: 'flex',
            gap: '1rem',
            padding: '1rem',
            border: '1px solid #e0e0e0',
            borderRadius: '8px',
          }}
        >
          {product.imageUrl && (
            <img
              src={product.imageUrl}
              alt={product.name ?? ''}
              style={{width: 80, height: 80, objectFit: 'cover'}}
            />
          )}
          <div>
            <h3 style={{margin: 0}}>
              <a
                href={product.clickUri ?? '#'}
                target="_blank"
                rel="noreferrer"
              >
                {product.name}
              </a>
            </h3>
            {product.brand && (
              <p style={{margin: '0.25rem 0', color: '#666'}}>
                {product.brand}
              </p>
            )}
            {product.price !== null && (
              <p style={{margin: '0.25rem 0', fontWeight: 'bold'}}>
                {formatPrice(product.price, product.currency)}
              </p>
            )}
            {product.description && (
              <p style={{margin: '0.25rem 0', fontSize: '0.9rem'}}>
                {product.description}
              </p>
            )}
          </div>
        </article>
      ))}
    </div>
  );
}

export default CommerceApp;

npm run build