"""DC Hub — Power Index demo + MCP server (Hugging Face Space). A live demo of the DC Hub Power Index (DCPI) that ALSO runs as an MCP server (launch(mcp_server=True)) so Hugging Face Agents / smolagents and any MCP client can query it. Seven tools: the two DCPI classics (dcpi_score, compare_markets) plus five bridge tools (grid scoreboard, facility search, market ranking, interconnection queue, hyperscaler deals) forwarded to DC Hub's canonical MCP server. The complete DC Hub MCP — every facility, market and deal in the index, plus real-time grid telemetry, fiber and gas — lives at https://dchub.cloud/mcp. Connect that for the full dataset (10 calls/day free anonymous; free key via its claim_free_key tool). ★ COUNTS ARE NOT TYPED IN THIS FILE. They resolve from DC Hub's canon endpoint (/api/v1/canon/phrases) via _canon() below, because a typed count here rots silently: this Space shipped a facilities figure ABOVE canon's citation-safe floor — an OVER-claim, the one direction a floored count must never take. Docstrings are the exception and state no count at all: they are the MCP tool schema, they are read once at import, and a network call in the import path is its own defect. Optional Space secret DCHUB_API_KEY (a dchub_ key) lifts the bridge tools to full-tier responses; without it they ride the anonymous free tier. UI NOTE: the 7 tool functions below are the MCP contract (names + docstrings become the tool schema at /gradio_api/mcp/schema) — keep them stable. UI-only events that reuse them (e.g. textbox .submit) pass api_name=False so they never mint a duplicate MCP tool. """ import json import os import time import uuid import requests import gradio as gr API = "https://dchub.cloud" FULL_MCP = "https://dchub.cloud/mcp" SPACE_MCP = "https://dchubcloud-dchub.hf.space/gradio_api/mcp/sse" TIMEOUT = 20 API_KEY = (os.environ.get("DCHUB_API_KEY") or "").strip() UA = "huggingface-space-dchub-bridge/1.0 (+https://dchub.cloud)" # ── Canon-derived public counts ──────────────────────────────────────────── # DC Hub publishes its own citation-safe floors at /api/v1/canon/phrases. They # are floored DOWN from the live measurement and are never above reality, which # is exactly the property a public, CC-BY-4.0 surface needs. # # Fail-open by construction: an unreadable canon yields "" for every phrase and # the copy renders COUNT-FREE. That asymmetry is deliberate — a missing number # is visible and self-correcting, a stale one is neither. _CANON_TTL = 3600 _CANON_TIMEOUT = 6 # short: this runs once at Space boot, before the UI _canon_cache = {"at": 0.0, "val": {}} def _canon(): """Live public floors: {'facilities','deals','markets','tools'} or {}.""" now = time.time() if _canon_cache["val"] and (now - _canon_cache["at"]) < _CANON_TTL: return _canon_cache["val"] try: r = requests.get(f"{API}/api/v1/canon/phrases", timeout=_CANON_TIMEOUT, headers={"User-Agent": UA}) d = r.json() if r.ok else {} val = {k: str(d[k]) for k in ("facilities", "deals", "markets", "tools") if d.get(k)} except Exception: val = {} if val: _canon_cache.update(at=now, val=val) return _canon_cache["val"] def _phrase(key, noun): """'20,400+ facilities', or '' when canon cannot be read. Callers must tolerate the empty string — that is the whole point. Build sentences so an absent phrase leaves them grammatical, never mangled. """ v = _canon().get(key) return f"{v} {noun}" if v else "" def _tools_clause(): """' + 83 tools', or '' — a clause that may vanish without breaking prose.""" p = _phrase("tools", "tools") return f" + {p}" if p else "" def _tools_strong(): """'83 tools, ' or '' — trailing comma travels with it.""" p = _phrase("tools", "tools") return f"{p}, " if p else "" def _verdict(score): try: s = float(score) except (TypeError, ValueError): return "—" if s >= 60: return "🟢 BUILD" if s >= 30: return "🟡 CAUTION" return "🔴 AVOID" def _slug(market): return (market or "").strip().lower().replace(",", "").replace(" ", "-") # ── DCPI reads ───────────────────────────────────────────────────────────── # /api/v1/dcpi/scores/ answers a keyless caller with a PREVIEW: the # BUILD / CAUTION / AVOID verdict and the market identity are free, but every # number (composite_score, avg_kwh_cents, queue_wait_months, ...) comes back # null with "_gated": true. Printing those nulls read as "None/100" and a "—" # verdict, and compare_markets coerced them to 0.0 and ranked every market # AVOID — Cheyenne (BUILD) included. So: use the API's own verdict when the # score is locked, say plainly that the number needs a key, and quote the # API's own _upgrade_cta rather than typing a price here. _VERDICT_BADGE = {"BUILD": "🟢 BUILD", "CAUTION": "🟡 CAUTION", "AVOID": "🔴 AVOID"} _VERDICT_ORDER = {"BUILD": 0, "CAUTION": 1, "AVOID": 2} def _dcpi_get(market): headers = {"User-Agent": UA} if API_KEY: headers["X-API-Key"] = API_KEY return requests.get(f"{API}/api/v1/dcpi/scores/{_slug(market)}", timeout=TIMEOUT, headers=headers) def _num(v): return v if isinstance(v, (int, float)) and not isinstance(v, bool) else None def _dcpi_verdict_word(d): """'BUILD' / 'CAUTION' / 'AVOID' from the score, else the API's verdict, else ''.""" score = _num(d.get("composite_score")) if score is not None: return _verdict(score).split()[-1] v = str(d.get("verdict") or "").strip().upper() return v if v in _VERDICT_BADGE else "" def _score_locked(d): return _num(d.get("composite_score")) is None def _key_note(d): cta = (d.get("_upgrade_cta") or "").strip() return ("The numeric score, power cost and time-to-power need a DC Hub API key" + (f". {cta}" if cta else ".") + f" Or connect the DC Hub MCP → **{FULL_MCP}**.") def dcpi_score(market: str) -> str: """Get the DC Hub Power Index (DCPI) for a U.S. data-center market: a 0-100 power-availability score with a BUILD / CAUTION / AVOID verdict, average power cost (cents/kWh), and modeled time-to-power (months). Answers "can I actually get power to build a data center in this market, and how soon?". Args: market: A U.S. market name, e.g. "Northern Virginia", "Phoenix", "Columbus", "Dallas", "Atlanta", "Cheyenne", "Omaha", "Tulsa". Without a DC Hub API key the verdict is returned and the numeric score is reported as needing a key (it is never printed as a number it isn't). """ try: r = _dcpi_get(market) if r.status_code == 404: return (f"No DCPI data for **{market}**. Try a major U.S. market — Northern " f"Virginia, Phoenix, Columbus, Dallas, Atlanta, Cheyenne, Omaha, Tulsa.") r.raise_for_status() d = r.json() name = d.get("market_name", market.title()) where = ", ".join(x for x in (d.get("state"), d.get("iso")) if x) head = f"### {name}" + (f" ({where})" if where else "") + " — DC Hub Power Index\n" if _score_locked(d): word = _dcpi_verdict_word(d) verdict_line = (f"- **Verdict: {_VERDICT_BADGE[word]}**\n" if word else "- Verdict: not in the keyless response\n") return ( head + verdict_line + f"- **Score: needs a key.** {_key_note(d)}\n\n" f"_DC Hub (dchub.cloud) · CC-BY-4.0._" ) score = d.get("composite_score") ttp = d.get("queue_wait_months") ttp_s = f"~{int(ttp)} months" if isinstance(ttp, (int, float)) else "n/a" return ( head + f"- **Verdict: {_verdict(score)}** · composite **{score}/100**\n" f"- Modeled time-to-power: **{ttp_s}**\n" f"- Avg power cost: **{d.get('avg_kwh_cents', '?')} ¢/kWh**\n" f"- Excess-power headroom: {d.get('excess_power_score', '?')}/100 · " f"grid constraint: {d.get('constraint_score', '?')}/100\n\n" f"_Modeled estimate from public ISO/EIA/queue data · DC Hub (dchub.cloud) · CC-BY-4.0._\n\n" f"_Full grid headroom, interconnection queue, fiber{_tools_clause()}: connect the DC Hub MCP → **{FULL_MCP}**._" ) except Exception as e: return f"DC Hub lookup failed ({e}). Query the full live MCP at {FULL_MCP}." def compare_markets(markets: str) -> str: """Rank several U.S. data-center markets by the DC Hub Power Index (DCPI), best power-availability (BUILD) first, to decide where to build. Args: markets: Comma-separated market names, e.g. "Northern Virginia, Phoenix, Cheyenne, Omaha". """ rows, locked = [], None for m in [x.strip() for x in (markets or "").split(",") if x.strip()][:8]: try: r = _dcpi_get(m) if r.status_code == 200: d = r.json() word = _dcpi_verdict_word(d) score = _num(d.get("composite_score")) if score is None and locked is None: locked = d # Verdict first, then score (locked scores sort last within a verdict). rows.append((_VERDICT_ORDER.get(word, 3), -(score if score is not None else -1), d.get("market_name", m.title()), word, score, d.get("avg_kwh_cents"), d.get("queue_wait_months"))) except Exception: pass if not rows: return "No DCPI data for those markets. Try major U.S. metros." rows.sort() out = ["### DCPI ranking — best power-availability first\n"] for _o, _s, name, word, score, cost, ttp in rows: badge = _VERDICT_BADGE.get(word, "verdict n/a") if score is None: out.append(f"- {badge} **{name}** — score needs a key") continue ttp_s = f"~{int(ttp)}mo" if isinstance(ttp, (int, float)) else "n/a" cost_s = f"{cost}¢/kWh" if _num(cost) is not None else "cost n/a" out.append(f"- {badge} **{name}** — {score}/100 · {cost_s} · {ttp_s} to power") if locked is not None: out.append("\n_Ranked by verdict: " + _key_note(locked).replace("**", "") + "_") out.append(f"\n_Full ranking across {_phrase('markets', 'markets') or 'every scored market'}: DC Hub MCP → {FULL_MCP}._") return "\n".join(out) # ── Bridge tools: forwarded to the canonical DC Hub MCP server ────────────── def _call_tool(name: str, arguments: dict) -> str: """Single-shot JSON-RPC tools/call against the canonical DC Hub MCP server.""" headers = { "Content-Type": "application/json", "Accept": "application/json, text/event-stream", "User-Agent": UA, "X-MCP-Platform": "huggingface", } if API_KEY: headers["X-API-Key"] = API_KEY body = { "jsonrpc": "2.0", "id": str(uuid.uuid4()), "method": "tools/call", "params": {"name": name, "arguments": {k: v for k, v in arguments.items() if v not in ("", None)}}, } try: r = requests.post(FULL_MCP, json=body, headers=headers, timeout=45) text = r.text if "data:" in text and text.lstrip().startswith("event:"): for line in text.splitlines(): if line.startswith("data:"): text = line[5:].strip() break payload = json.loads(text) result = payload.get("result") or payload content = result.get("content") if isinstance(content, list): out = "\n".join(c.get("text", "") for c in content if c.get("type") == "text") else: out = json.dumps(result, indent=2) try: return json.dumps(json.loads(out), indent=2) except Exception: return out except Exception as e: return json.dumps({"error": str(e)[:300], "hint": f"DC Hub MCP unreachable — retry, or connect {FULL_MCP} directly"}) def grid_scoreboard() -> str: """Live ranked scoreboard of power grids (7 US ISOs + Great Britain, EU zones, Taiwan, Japan, South Korea, Brazil, and more): fuel mix, renewable share, and demand right now. Greenest grids first. No parameters.""" return _call_tool("get_grid_scoreboard", {}) def search_facilities(query: str = "", country: str = "", state: str | None = None, operator: str | None = None, min_capacity_mw: float | None = None, limit: int | None = 10) -> str: """Search DC Hub's global index of data-center facilities (many independently verified) across 170+ countries. Args: query: Free-text search over facility name/operator/location. country: ISO 3166-1 alpha-2 country code, e.g. US, GB, SG. state: US state abbreviation, e.g. VA, TX. operator: Operator/provider name, e.g. Equinix, Digital Realty. min_capacity_mw: Minimum power capacity in megawatts. limit: Max results (1-500). Every argument is optional; pass only the filters you need. """ # Gradio marks an MCP/API argument required unless its UI component has an # initial value or the Python default is None. State, Operator and Min # capacity had neither, so a call with just {"query": ...} failed with # "No value provided for required argument: state". The None defaults above # are what make them optional in /gradio_api/mcp/schema. return _call_tool("search_facilities", { "query": query, "country": country, "state": state, "operator": operator, "min_capacity_mw": min_capacity_mw or None, "limit": int(limit or 0) or 10}) def rank_markets(criteria: str = "most_capacity", region: str = "us", limit: int = 10) -> str: """Rank data-center markets by a live criterion from the DC Hub Power Index. Args: criteria: One of cheapest_power, most_capacity, most_operators, fastest. region: global, us, canada, eu, apac, or americas. limit: Number of markets to return (1-50). """ return _call_tool("rank_markets", {"criteria": criteria, "region": region, "limit": int(limit) or 10}) def interconnection_queue(iso: str = "") -> str: """US grid interconnection-queue intelligence — how much capacity is waiting to connect, by ISO/RTO. Args: iso: Optional drill-down: ERCOT, PJM, MISO, CAISO, SPP, NYISO, or ISONE. Empty = national overview. """ return _call_tool("get_interconnection_queue", {"iso": iso}) def hyperscaler_deals(limit: int = 10) -> str: """Recent hyperscaler / AI-capex deals from DC Hub's deal tracker. Args: limit: Number of recent deals to return. """ return _call_tool("hyperscaler_deals", {"limit": int(limit) or 10}) # ── UI ────────────────────────────────────────────────────────────────────── # Everything below is presentation only — no new MCP tools are minted here # (UI-only events pass api_name=False). CSS = """ .gradio-container { max-width: 1080px !important; margin: 0 auto !important; } .dc-hero { border-radius: 14px; padding: 26px 28px 22px; margin-bottom: 4px; background: linear-gradient(120deg, #0b1220 0%, #101c33 55%, #1a2a17 100%); border: 1px solid rgba(250, 204, 21, .25); } .dc-hero h1 { margin: 0 0 6px; font-size: 1.65rem; line-height: 1.2; color: #f8fafc; } .dc-hero h1 .amp { color: #facc15; } .dc-hero .sub { margin: 0 0 14px; font-size: 1rem; color: #cbd5e1; max-width: 62rem; } .dc-hero .sub a { color: #facc15; text-decoration: none; } .dc-chips { display: flex; flex-wrap: wrap; gap: 8px; } .dc-chip { font-size: .8rem; padding: 4px 11px; border-radius: 999px; background: rgba(148, 163, 184, .14); border: 1px solid rgba(148, 163, 184, .28); color: #e2e8f0; white-space: nowrap; } .dc-chip.build { background: rgba(34, 197, 94, .14); border-color: rgba(34, 197, 94, .45); color: #86efac; } .dc-chip.caution { background: rgba(234, 179, 8, .14); border-color: rgba(234, 179, 8, .45); color: #fde047; } .dc-chip.avoid { background: rgba(239, 68, 68, .14); border-color: rgba(239, 68, 68, .45); color: #fca5a5; } .dc-foot { text-align: center; font-size: .85rem; opacity: .8; margin-top: 6px; } .dc-mcp-note { font-size: .9rem; margin: 2px 0 8px; } """ def _hero(): """Built when the UI is constructed, so canon is read once per boot.""" _mk = _phrase("markets", "markets") _fa = _phrase("facilities", "facilities") _dl = _phrase("deals", "tracked deals") _chips = "".join( f'{c}' for c in (_mk, _fa, "live grid telemetry", _dl, "CC-BY-4.0") if c) return f"""

⚡ DC Hub — Power Index

Can you actually get power to build a data center here? Live 0–100 power-availability scores for data-center markets worldwide, with a verdict, power cost, and modeled time-to-power — from dchub.cloud.

🟢 BUILD ≥ 60 🟡 CAUTION 30–59 🔴 AVOID < 30 {_chips}
""" def _claude_code_snippet(): _t = _phrase("tools", "tools") _full = f"({_t}; 10 calls/day free, no key needed)" if _t else "(10 calls/day free, no key needed)" return f"""# This Space (7 tools, free) claude mcp add --transport sse dchub-power-index {SPACE_MCP} # Full DC Hub MCP {_full} claude mcp add --transport http dchub {FULL_MCP}""" JSON_SNIPPET = f"""{{ "mcpServers": {{ "dchub-power-index": {{ "url": "{SPACE_MCP}" }}, "dchub": {{ "url": "{FULL_MCP}" }} }} }}""" SMOLAGENTS_SNIPPET = f"""from smolagents import CodeAgent, InferenceClientModel from smolagents.mcp_client import MCPClient with MCPClient({{"url": "{SPACE_MCP}"}}) as tools: agent = CodeAgent(tools=tools, model=InferenceClientModel()) agent.run("Which U.S. market should I build a 100MW data center in?")""" # DeepSeek Harness (deepseek-ai/deepseek-harness) — its MCP client plugin is # @deepseek-ai/dsh-mcp-client; transport streamable-http takes url + headers. # ★ Points at FULL_MCP, not this Space: the harness can hold a key, so there is # no reason to route it through a 7-tool bridge. DeepSeek's CONSUMER chat has # no MCP UI at all — that is why this tab names the harness, not "DeepSeek". DEEPSEEK_SNIPPET = f"""- id: mcp-dchub name: '@deepseek-ai/dsh-mcp-client' config: serverName: dchub transport: streamable-http url: {FULL_MCP}""" with gr.Blocks(title="DC Hub — Power Index", theme=gr.themes.Soft(primary_hue="amber", neutral_hue="slate"), css=CSS) as demo: gr.HTML(_hero()) with gr.Accordion("🔌 Connect via MCP — Claude Code · Claude Desktop · Cursor · smolagents", open=False): gr.Markdown( f"

This Space is an MCP server " f"(7 tools) at {SPACE_MCP}. The full DC Hub MCP — " f"{_tools_strong()}real-time grid telemetry, fiber, gas, " f"water, tax incentives, deal intelligence — is at " f"{FULL_MCP} (10 calls/day free anonymous; mint a free " f"key with its claim_free_key tool, or try it with zero " f"setup at dchub.cloud/playground).

") gr.Markdown( f"

Per-client setup recipes — Claude, Cursor, " f"VS Code, Cline, Windsurf, ChatGPT, Gemini, Perplexity, Grok, " f"DeepSeek Harness — are at " f"dchub.cloud/connect, and the same list " f"in agent form at dchub.cloud/ai-agents.

") with gr.Tabs(): with gr.Tab("Claude Code"): gr.Code(_claude_code_snippet(), language="shell", label="terminal") with gr.Tab("Desktop · Cursor"): gr.Code(JSON_SNIPPET, language="json", label="mcp config (JSON) — any MCP client") with gr.Tab("DeepSeek Harness"): gr.Code(DEEPSEEK_SNIPPET, language="yaml", label="harness profile YAML — consumer DeepSeek chat has no MCP UI") with gr.Tab("smolagents"): gr.Code(SMOLAGENTS_SNIPPET, language="python", label="python — HF Agents / smolagents") with gr.Tab("⚡ Score a market"): inp = gr.Textbox(label="U.S. data-center market", value="Northern Virginia", placeholder="Northern Virginia, Phoenix, Columbus, Cheyenne…", info="One market name — get the verdict, score, cost, and time-to-power.") btn = gr.Button("Get DCPI score", variant="primary") out = gr.Markdown() btn.click(dcpi_score, inp, out) inp.submit(dcpi_score, inp, out, api_name=False) gr.Examples([["Northern Virginia"], ["Phoenix"], ["Columbus"], ["Cheyenne"], ["Dallas"], ["Omaha"]], inp) with gr.Tab("🏆 Compare markets"): cinp = gr.Textbox(label="Markets (comma-separated)", value="Northern Virginia, Phoenix, Cheyenne, Omaha, Dallas", info="Up to 8 U.S. markets — ranked best power-availability first.") cbtn = gr.Button("Rank by power availability", variant="primary") cout = gr.Markdown() cbtn.click(compare_markets, cinp, cout) cinp.submit(compare_markets, cinp, cout, api_name=False) gr.Examples([["Northern Virginia, Phoenix, Cheyenne, Omaha, Dallas"], ["Atlanta, Columbus, Tulsa"], ["Dallas, Austin, San Antonio"]], cinp) with gr.Tab("🌍 Grid scoreboard"): gr.Markdown("Live ranked grids — US ISOs + GB, EU, Taiwan, Japan, South Korea, " "Brazil: fuel mix, renewable share, and demand **right now**. " "Greenest first.") gout = gr.Code(language="json", label="live scoreboard") gr.Button("Live grid scoreboard (greenest first)", variant="primary").click( grid_scoreboard, [], gout) with gr.Tab("🔎 Search facilities"): with gr.Row(): s_q = gr.Textbox(label="Query", value="Ashburn", placeholder="name / operator / location") s_c = gr.Textbox(label="Country (ISO-2)", value="US", placeholder="US, GB, SG…") s_s = gr.Textbox(label="State", placeholder="VA, TX…") with gr.Row(): s_o = gr.Textbox(label="Operator", placeholder="Equinix, Digital Realty…") s_m = gr.Number(label="Min capacity (MW)", value=0) s_l = gr.Number(label="Limit", value=10) sout = gr.Code(language="json", label="facilities") gr.Button("Search facilities", variant="primary").click( search_facilities, [s_q, s_c, s_s, s_o, s_m, s_l], sout) with gr.Tab("📊 Rank markets"): with gr.Row(): r_c = gr.Dropdown(["most_capacity", "cheapest_power", "most_operators", "fastest"], value="most_capacity", label="Criteria") r_r = gr.Dropdown(["us", "global", "canada", "eu", "apac", "americas"], value="us", label="Region") r_l = gr.Number(label="Limit", value=10) rout = gr.Code(language="json", label="ranking") gr.Button("Rank markets", variant="primary").click( rank_markets, [r_c, r_r, r_l], rout) with gr.Tab("🔌 Interconnection queue"): q_i = gr.Dropdown(["", "ERCOT", "PJM", "MISO", "CAISO", "SPP", "NYISO", "ISONE"], value="", label="ISO", info="How much capacity is waiting to connect. Empty = national overview.") qout = gr.Code(language="json", label="queue intelligence") gr.Button("Queue intelligence", variant="primary").click( interconnection_queue, [q_i], qout) with gr.Tab("💸 Hyperscaler deals"): d_l = gr.Number(label="Limit", value=10, info="Most recent hyperscaler / AI-capex deals from the DC Hub tracker.") dout = gr.Code(language="json", label="deals") gr.Button("Recent AI-capex deals", variant="primary").click( hyperscaler_deals, [d_l], dout) gr.HTML( f"
Data: DC Hub, CC-BY-4.0 — " f"the live infrastructure data layer for AI agents · " f"playground · " f"free key · " f"connect · " f"ai-agents
") if __name__ == "__main__": # mcp_server=True exposes all 7 tools as MCP tools at /gradio_api/mcp/sse — # usable by HF Agents, smolagents, and any MCP client. demo.launch(mcp_server=True)