"""
Dexcom glucose + carb search MCP server for Smithery deployment.
"""
import json
from os import getenv
from urllib.parse import quote_plus
import httpx
from mcp.server.fastmcp import Context, FastMCP
from pydantic import BaseModel, Field
from pydexcom import Dexcom
from smithery.decorators import smithery
class ConfigSchema(BaseModel):
"""Session-level configuration supplied by Smithery."""
username: str = Field(..., description="Dexcom username/email")
password: str = Field(..., description="Dexcom password")
region: str = Field(
...,
description='Dexcom region code ("us" for US servers, "ous" for outside US)',
)
brightdata_api_key: str = Field(..., description="Bright Data API key for search")
@smithery.server(config_schema=ConfigSchema)
def create_server():
"""Create and configure the MCP server."""
server = FastMCP("Dexcom + Carb Search")
@server.tool()
def read_blood_glucose(
ctx: Context,
username: str | None = None,
password: str | None = None,
region: str | None = None,
) -> str:
"""
Fetch the latest blood glucose reading from Dexcom.
Credentials/region are pulled from Smithery session config by default,
but can be overridden per-call via arguments.
"""
session_config = getattr(ctx, "session_config", None)
dex_username = (
username
or getattr(session_config, "username", None)
or getenv("DEXCOM_USERNAME")
)
dex_password = (
password
or getattr(session_config, "password", None)
or getenv("DEXCOM_PASSWORD")
)
dex_region = (
region or getattr(session_config, "region", None) or getenv("DEXCOM_REGION")
)
dex_region = dex_region.lower() if dex_region else None
if not dex_username or not dex_password or not dex_region:
return (
"Missing Dexcom credentials. Provide username/password/region arguments or "
"configure them in the Smithery session settings."
)
try:
dexcom = Dexcom(
username=dex_username,
password=dex_password,
region=dex_region,
)
reading = dexcom.get_current_glucose_reading()
except Exception as exc: # noqa: BLE001
return f"Failed to fetch glucose reading: {exc}"
if not reading:
return "No glucose reading available right now."
mg_dl = getattr(reading, "mg_dl", None)
trend = getattr(reading, "trend_description", None)
if mg_dl is None:
return f"Current glucose reading: {reading}"
if trend:
return f"Current glucose: {mg_dl} mg/dL (trend: {trend})."
return f"Current glucose: {mg_dl} mg/dL."
@server.tool()
def search_carbs(
ctx: Context,
food_name: str,
) -> str:
"""
Search Google via Bright Data for carb info.
The query is forced to: "how many carbs in <food_name>". Returns cleaned
organic results (JSON). Engine is fixed to Google; first page only.
"""
session_config = getattr(ctx, "session_config", None)
brightdata_api_key = (
getattr(session_config, "brightdata_api_key", None)
or getenv("BRIGHTDATA_API_KEY")
)
if not brightdata_api_key:
return (
"Missing Bright Data API key. Configure brightdata_api_key in the "
"Smithery session settings or set BRIGHTDATA_API_KEY."
)
query = f"how many carbs in {food_name}"
search_url = f"https://www.google.com/search?q={quote_plus(query)}&start=0"
try:
response = httpx.post(
"https://api.brightdata.com/request",
json={
"url": search_url,
"format": "raw",
"data_format": "parsed_light",
"zone": "mcp_unlocker",
},
headers={"Authorization": f"Bearer {brightdata_api_key}"},
timeout=30,
)
response.raise_for_status()
except Exception as exc: # noqa: BLE001
return f"Search request failed: {exc}"
try:
data = response.json()
except Exception:
return response.text
organic = data.get("organic", []) if isinstance(data, dict) else []
cleaned: list[dict[str, str]] = []
for entry in organic:
if not isinstance(entry, dict):
continue
link = str(entry.get("link", "")).strip()
title = str(entry.get("title", "")).strip()
description = str(entry.get("description", "")).strip()
if not link or not title:
continue
cleaned.append({"link": link, "title": title, "description": description})
return json.dumps({"organic": cleaned}, indent=2)
return server
