from __future__ import annotations
import html
import os
from dataclasses import dataclass
from typing import Any
import gradio as gr
from agent import (
DraftRecommendationRequest,
DraftTeam,
recommend_draft_action,
warm_static_dota_data,
)
from mcp import initialize_opendota_mcp_server
@dataclass(frozen=True)
class DraftStep:
number: int
phase: str
action: str
slot: int
team: str
DRAFT_SEQUENCE = [
DraftStep(1, "Phase 1", "Ban", 1, "A"),
DraftStep(2, "Phase 1", "Ban", 2, "B"),
DraftStep(3, "Phase 1", "Ban", 3, "A"),
DraftStep(4, "Phase 1", "Ban", 4, "B"),
DraftStep(5, "Phase 1", "Pick", 1, "A"),
DraftStep(6, "Phase 1", "Pick", 2, "B"),
DraftStep(7, "Phase 1", "Pick", 3, "B"),
DraftStep(8, "Phase 1", "Pick", 4, "A"),
DraftStep(9, "Phase 2", "Ban", 5, "A"),
DraftStep(10, "Phase 2", "Ban", 6, "B"),
DraftStep(11, "Phase 2", "Ban", 7, "A"),
DraftStep(12, "Phase 2", "Ban", 8, "B"),
DraftStep(13, "Phase 2", "Pick", 5, "A"),
DraftStep(14, "Phase 2", "Pick", 6, "B"),
DraftStep(15, "Phase 2", "Pick", 7, "B"),
DraftStep(16, "Phase 2", "Pick", 8, "A"),
DraftStep(17, "Phase 3", "Ban", 9, "A"),
DraftStep(18, "Phase 3", "Ban", 10, "B"),
DraftStep(19, "Phase 3", "Pick", 9, "A"),
DraftStep(20, "Phase 3", "Pick", 10, "B"),
]
HERO_SUGGESTIONS = [
"Abaddon",
"Alchemist",
"Ancient Apparition",
"Anti-Mage",
"Arc Warden",
"Axe",
"Bane",
"Batrider",
"Beastmaster",
"Bloodseeker",
"Bounty Hunter",
"Brewmaster",
"Bristleback",
"Broodmother",
"Centaur Warrunner",
"Chaos Knight",
"Chen",
"Clinkz",
"Clockwerk",
"Crystal Maiden",
"Dark Seer",
"Dark Willow",
"Dawnbreaker",
"Dazzle",
"Death Prophet",
"Disruptor",
"Doom",
"Dragon Knight",
"Drow Ranger",
"Earth Spirit",
"Earthshaker",
"Elder Titan",
"Ember Spirit",
"Enchantress",
"Enigma",
"Faceless Void",
"Grimstroke",
"Gyrocopter",
"Hoodwink",
"Huskar",
"Invoker",
"Io",
"Jakiro",
"Juggernaut",
"Keeper of the Light",
"Kez",
"Kunkka",
"Largo",
"Legion Commander",
"Leshrac",
"Lich",
"Lifestealer",
"Lina",
"Lion",
"Lone Druid",
"Luna",
"Lycan",
"Magnus",
"Marci",
"Mars",
"Medusa",
"Meepo",
"Mirana",
"Monkey King",
"Muerta",
"Naga Siren",
"Nature's Prophet",
"Necrophos",
"Night Stalker",
"Nyx Assassin",
"Ogre Magi",
"Omniknight",
"Oracle",
"Outworld Destroyer",
"Pangolier",
"Phantom Assassin",
"Phantom Lancer",
"Phoenix",
"Primal Beast",
"Puck",
"Pudge",
"Pugna",
"Queen of Pain",
"Razor",
"Riki",
"Ringmaster",
"Rubick",
"Sand King",
"Shadow Demon",
"Shadow Fiend",
"Shadow Shaman",
"Silencer",
"Skywrath Mage",
"Slardar",
"Slark",
"Snapfire",
"Sniper",
"Spectre",
"Spirit Breaker",
"Storm Spirit",
"Sven",
"Techies",
"Templar Assassin",
"Terrorblade",
"Tidehunter",
"Timbersaw",
"Tinker",
"Tiny",
"Treant Protector",
"Troll Warlord",
"Tusk",
"Underlord",
"Undying",
"Ursa",
"Vengeful Spirit",
"Venomancer",
"Viper",
"Visage",
"Void Spirit",
"Warlock",
"Weaver",
"Windranger",
"Winter Wyvern",
"Witch Doctor",
"Wraith King",
"Zeus",
]
def new_state() -> dict[str, Any]:
opendota_api_key = os.getenv("OPENDOTA_API_KEY", "")
return {
"api_token": opendota_api_key,
"api_token_configured": bool(opendota_api_key),
"initialized": False,
"team_names": {"A": "Team A", "B": "Team B"},
"user_team": "A",
"opponent_team": "B",
"sides": {"A": "Radiant", "B": "Dire"},
"athlete": "",
"player_ids": {"A": [], "B": []},
"history": [],
"insights": [],
}
def preserve_token(source: dict[str, Any] | None, target: dict[str, Any]) -> dict[str, Any]:
if source:
target["api_token"] = source.get("api_token", "") or os.getenv("OPENDOTA_API_KEY", "")
target["api_token_configured"] = bool(target["api_token"])
return target
def team_label(state: dict[str, Any], team_key: str) -> str:
marker = "You" if team_key == state["user_team"] else "Opponent"
return f'{state["team_names"][team_key]} ({marker}, {state["sides"][team_key]})'
def current_step(state: dict[str, Any]) -> DraftStep | None:
index = len(state["history"])
if index >= len(DRAFT_SEQUENCE):
return None
return DRAFT_SEQUENCE[index]
def build_state(
user_team_name: str,
opponent_team_name: str,
user_side: str,
first_drafter: str,
opponent_athlete: str,
user_player_ids: str,
opponent_player_ids: str,
existing_state: dict[str, Any] | None = None,
) -> dict[str, Any]:
user_key = "A" if first_drafter == "You draft first" else "B"
opponent_key = "B" if user_key == "A" else "A"
opponent_side = "Dire" if user_side == "Radiant" else "Radiant"
names = {
user_key: user_team_name.strip() or "Your Team",
opponent_key: opponent_team_name.strip() or "Opponent",
}
sides = {user_key: user_side, opponent_key: opponent_side}
state = preserve_token(existing_state, new_state())
state.update(
{
"initialized": True,
"team_names": names,
"user_team": user_key,
"opponent_team": opponent_key,
"sides": sides,
"athlete": opponent_athlete.strip(),
"player_ids": {
user_key: parse_player_ids(user_player_ids),
opponent_key: parse_player_ids(opponent_player_ids),
},
}
)
return state
def parse_player_ids(raw_ids: str) -> list[str]:
return [
item.strip()
for item in raw_ids.replace("\n", ",").split(",")
if item.strip()
]
def build_timeline(state: dict[str, Any]) -> list[list[str]]:
rows = []
for index, step in enumerate(DRAFT_SEQUENCE):
entry = state["history"][index] if index < len(state["history"]) else None
status = "Done" if entry else "Current" if index == len(state["history"]) else "Pending"
hero = entry["hero"] if entry else ""
rows.append(
[
str(step.number),
step.phase,
step.action,
team_label(state, step.team),
hero,
status,
]
)
return rows
def _render_slots(heroes: list[str]) -> str:
cells = []
for index in range(5):
if index < len(heroes):
cells.append(f'{html.escape(heroes[index])}')
else:
cells.append('')
return "".join(cells)
def build_board(state: dict[str, Any]) -> str:
sections = []
for key in ["A", "B"]:
picks = [
item["hero"]
for item in state["history"]
if item["team"] == key and item["action"] == "Pick"
]
bans = [
item["hero"]
for item in state["history"]
if item["team"] == key and item["action"] == "Ban"
]
owner = "you" if key == state.get("user_team") else "opponent"
sections.append(
f"""
{html.escape(team_label(state, key))}
Picks
{_render_slots(picks)}
Bans
{_render_slots(bans)}
"""
)
return f'{"".join(sections)}
'
def next_prompt(state: dict[str, Any]) -> str:
if not state["initialized"]:
return "Initialize the draft to begin."
step = current_step(state)
if step is None:
return "Draft complete. Review the final picks, bans, and backend insight log."
return (
f"Step {step.number}/20 - {step.phase}: {step.action} {step.slot} for "
f"{team_label(state, step.team)}"
)
def heroes_by_team_and_action(
state: dict[str, Any], team_key: str, action: str
) -> list[str]:
return [
item["hero"]
for item in state["history"]
if item["team"] == team_key and item["action"] == action
]
def team_draft_state(state: dict[str, Any], team_key: str) -> dict[str, Any]:
return {
"team": state["team_names"][team_key],
"side": state["sides"][team_key],
"picks": heroes_by_team_and_action(state, team_key, "Pick"),
"bans": heroes_by_team_and_action(state, team_key, "Ban"),
"player_ids": state.get("player_ids", {}).get(team_key, []),
}
def make_backend_payload(state: dict[str, Any]) -> dict[str, Any]:
user_key = state["user_team"]
opponent_key = state["opponent_team"]
return {
"event": "draft_state_update",
"draft_position": len(state["history"]),
"complete": current_step(state) is None,
"you": team_draft_state(state, user_key),
"opponent": {
**team_draft_state(state, opponent_key),
"athlete": state["athlete"] or "unknown",
},
"opendota_api_token_configured": bool(state.get("api_token_configured")),
}
def sync_backend_payload(state: dict[str, Any]) -> dict[str, Any]:
state["last_payload"] = make_backend_payload(state) if state.get("initialized") else {}
return state
def insight_marker(position: int) -> str:
return f""
def mark_insight(position: int, text: str) -> str:
return f"{insight_marker(position)}\n{text}"
def remove_insights_for_position(state: dict[str, Any], position: int) -> None:
marker = insight_marker(position)
insights = state.get("insights", [])
marked = [insight for insight in insights if insight.startswith(marker)]
if marked:
state["insights"] = [
insight for insight in insights if not insight.startswith(marker)
]
return
remove_count = 1
if recommendation_request_from_state(state) is not None:
remove_count += 1
del insights[:remove_count]
def hero_key(hero: str) -> str:
return "".join(char for char in hero.lower() if char.isalnum())
def hero_is_unavailable(state: dict[str, Any], hero: str) -> bool:
normalized = hero_key(hero)
return any(
hero_key(item.get("hero", "")) == normalized
for item in state.get("history", [])
)
def placeholder_insight(payload: dict[str, Any]) -> str:
opponent = payload["opponent"]
athlete = opponent["athlete"]
athlete_text = (
f" for athlete `{athlete}`"
if athlete and athlete != "unknown"
else ""
)
return (
f"Queued backend lookup for the current draft state{athlete_text}.\n\n"
"Backend draft context:\n"
f"- Your picks: {', '.join(payload['you']['picks']) or 'none'}.\n"
f"- Your bans: {', '.join(payload['you']['bans']) or 'none'}.\n"
f"- Opponent picks: {', '.join(opponent['picks']) or 'none'}.\n"
f"- Opponent bans: {', '.join(opponent['bans']) or 'none'}.\n\n"
"Temporary UI guidance: treat this panel as the actionable recommendation surface "
"that will be replaced by generated insights."
)
def recommendation_request_from_state(state: dict[str, Any]) -> DraftRecommendationRequest | None:
step = current_step(state)
if not step or step.team != state["user_team"]:
return None
user_key = state["user_team"]
opponent_key = state["opponent_team"]
unavailable_heroes = [
item["hero"]
for item in state["history"]
if item.get("hero")
]
return DraftRecommendationRequest(
action=step.action, # type: ignore[arg-type]
step_number=step.number,
phase=step.phase,
slot=step.slot,
you=DraftTeam(
name=state["team_names"][user_key],
side=state["sides"][user_key],
picks=heroes_by_team_and_action(state, user_key, "Pick"),
bans=heroes_by_team_and_action(state, user_key, "Ban"),
player_ids=state.get("player_ids", {}).get(user_key, []),
),
opponent=DraftTeam(
name=state["team_names"][opponent_key],
side=state["sides"][opponent_key],
picks=heroes_by_team_and_action(state, opponent_key, "Pick"),
bans=heroes_by_team_and_action(state, opponent_key, "Ban"),
player_ids=state.get("player_ids", {}).get(opponent_key, []),
),
unavailable_heroes=unavailable_heroes,
opendota_api_key=state.get("api_token", ""),
)
async def add_recommendation_if_user_turn(state: dict[str, Any]) -> dict[str, Any]:
request = recommendation_request_from_state(state)
if not request:
return state
recommendation = await recommend_draft_action(request)
state["insights"].insert(0, mark_insight(len(state["history"]), recommendation))
return state
def render_outputs(state: dict[str, Any], selected_hero: str | None = ""):
prompt = next_prompt(state)
board = build_board(state)
timeline = build_timeline(state) if state["initialized"] else []
insight = "\n\n---\n\n".join(state["insights"]) or (
"Backend updates will appear here as soon as you submit a pick or ban."
)
payload = state.get("last_payload", {})
submit_enabled = bool(state["initialized"] and current_step(state))
return (
state,
prompt,
gr.update() if selected_hero is None else selected_hero,
board,
timeline,
insight,
payload,
gr.update(interactive=submit_enabled),
gr.update(interactive=bool(state["history"])),
)
THINKING_INSIGHT = "⏳ _Analyzing the current draft and generating a recommendation…_"
async def stream_with_recommendation(
state: dict[str, Any],
selected_hero: str = "",
extra: tuple[Any, ...] = (),
):
"""Render the draft surface immediately, then stream in the recommendation.
The agent call (OpenAI + OpenDota MCP) can take several seconds, so we yield
the updated board/timeline first and re-yield once the recommendation lands
instead of blocking the UI on the whole agent loop.
"""
pending = recommendation_request_from_state(state) is not None
snapshot = list(render_outputs(state, selected_hero))
if pending:
existing = state["insights"]
snapshot[5] = (
f"{THINKING_INSIGHT}\n\n---\n\n" + "\n\n---\n\n".join(existing)
if existing
else THINKING_INSIGHT
)
yield (*snapshot, *extra)
if pending:
await add_recommendation_if_user_turn(state)
yield (*render_outputs(state, None), *extra)
async def start_draft(
api_token: str,
user_team_name: str,
opponent_team_name: str,
user_side: str,
first_drafter: str,
opponent_athlete: str,
user_player_ids: str,
opponent_player_ids: str,
existing_state: dict[str, Any],
):
seeded = dict(existing_state) if existing_state else new_state()
clean_token = (api_token or "").strip() or os.getenv("OPENDOTA_API_KEY", "")
seeded["api_token"] = clean_token
seeded["api_token_configured"] = bool(clean_token)
state = build_state(
user_team_name,
opponent_team_name,
user_side,
first_drafter,
opponent_athlete,
user_player_ids,
opponent_player_ids,
seeded,
)
sync_backend_payload(state)
async for chunk in stream_with_recommendation(
state,
extra=(gr.update(visible=False), gr.update(visible=True)),
):
yield chunk
async def submit_hero(hero: str, state: dict[str, Any]):
if not state or not state.get("initialized"):
state = preserve_token(state, new_state())
yield render_outputs(state, hero.strip())
return
step = current_step(state)
clean_hero = hero.strip()
if step is None or not clean_hero:
yield render_outputs(state, clean_hero)
return
if hero_is_unavailable(state, clean_hero):
yield render_outputs(state, clean_hero)
return
entry = {
"step": step.number,
"phase": step.phase,
"action": step.action,
"slot": str(step.slot),
"team": step.team,
"hero": clean_hero,
}
state["history"].append(entry)
sync_backend_payload(state)
state["insights"].insert(
0,
mark_insight(len(state["history"]), placeholder_insight(state["last_payload"])),
)
async for chunk in stream_with_recommendation(state):
yield chunk
async def undo_last(state: dict[str, Any]):
if not state:
state = new_state()
if not state.get("history"):
yield render_outputs(state)
return
undone_position = len(state["history"])
remove_insights_for_position(state, undone_position)
state["history"].pop()
sync_backend_payload(state)
async for chunk in stream_with_recommendation(state):
yield chunk
async def reset_draft(state: dict[str, Any]):
if not state or not state.get("initialized"):
state = preserve_token(state, new_state())
sync_backend_payload(state)
yield render_outputs(state)
return
state["history"] = []
state["insights"] = []
sync_backend_payload(state)
async for chunk in stream_with_recommendation(state):
yield chunk
CSS = """
.gradio-container {
max-width: 1240px !important;
}
/* Page header */
.app-header h1 {
font-size: 28px;
font-weight: 700;
line-height: 1.2;
margin: 0 0 4px;
color: var(--body-text-color);
}
.app-header p {
margin: 0;
font-size: 15px;
color: var(--body-text-color-subdued);
}
/* Setup page: centered, narrow */
.setup-page {
max-width: 820px;
margin: 24px auto;
}
/* Section subtitle */
.section-title h3 {
margin: 0 0 2px;
font-size: 16px;
font-weight: 600;
color: var(--body-text-color);
}
/* Current-step banner */
.action-banner {
border: 1px solid var(--border-color-primary);
border-left: 4px solid var(--primary-500);
border-radius: var(--radius-lg);
background: var(--background-fill-secondary);
padding: 14px 18px !important;
}
.action-banner p {
margin: 0;
font-size: 16px;
font-weight: 600;
color: var(--body-text-color);
}
/* Draft board */
.draft-board {
display: grid;
grid-template-columns: repeat(2, minmax(0, 1fr));
gap: 16px;
}
.draft-panel {
border: 1px solid var(--border-color-primary);
border-radius: var(--radius-lg);
padding: 16px;
background: var(--background-fill-secondary);
}
.draft-panel.you {
border-top: 3px solid var(--primary-500);
}
.draft-panel.opponent {
border-top: 3px solid var(--neutral-400);
}
.draft-panel h3 {
font-size: 15px;
font-weight: 700;
margin: 0 0 8px;
color: var(--body-text-color);
}
.slot-label {
font-size: 11px;
font-weight: 700;
letter-spacing: 0.05em;
text-transform: uppercase;
color: var(--body-text-color-subdued);
margin: 12px 0 6px;
}
.slot-grid {
display: grid;
grid-template-columns: repeat(5, minmax(0, 1fr));
gap: 6px;
}
.slot-grid span {
min-height: 44px;
border-radius: var(--radius-sm);
padding: 6px;
display: flex;
align-items: center;
justify-content: center;
text-align: center;
font-size: 12px;
line-height: 1.2;
overflow-wrap: anywhere;
color: var(--body-text-color);
background: var(--background-fill-primary);
border: 1px dashed var(--border-color-primary);
}
.slot-grid span.filled {
border-style: solid;
font-weight: 600;
}
.slot-grid.picks span.filled {
background: rgba(34, 197, 94, 0.16);
border-color: rgba(34, 197, 94, 0.55);
}
.slot-grid.bans span.filled {
background: rgba(239, 68, 68, 0.14);
border-color: rgba(239, 68, 68, 0.5);
}
@media (max-width: 760px) {
.draft-board {
grid-template-columns: 1fr;
}
.slot-grid {
grid-template-columns: repeat(2, minmax(0, 1fr));
}
}
"""
THEME = gr.themes.Soft(
primary_hue="red",
neutral_hue="slate",
font=[gr.themes.GoogleFont("Inter"), "ui-sans-serif", "system-ui", "sans-serif"],
)
warm_static_dota_data(HERO_SUGGESTIONS)
initialize_opendota_mcp_server()
with gr.Blocks(title="Dota 2 Draft Assistant") as demo:
state = gr.State(new_state())
with gr.Column(elem_classes=["setup-page"]) as setup_page:
gr.HTML(
"""
"""
)
with gr.Group():
gr.Markdown("### Teams & draft order", elem_classes=["section-title"])
with gr.Row():
user_team_name = gr.Textbox(
label="Your team",
value="Your Team",
max_lines=1,
)
opponent_team_name = gr.Textbox(
label="Opponent team",
value="Opponent",
max_lines=1,
)
with gr.Row():
user_side = gr.Radio(
["Radiant", "Dire"],
value="Radiant",
label="Your side",
)
first_drafter = gr.Radio(
["You draft first", "Opponent drafts first"],
value="You draft first",
label="Draft order",
)
opponent_athlete = gr.Textbox(
label="Opponent esports athlete",
placeholder="Optional player handle for later stat lookup",
max_lines=1,
)
with gr.Row():
user_player_ids = gr.Textbox(
label="Your Steam player IDs",
placeholder="Optional, comma-separated for future player stat weighting",
max_lines=2,
)
opponent_player_ids = gr.Textbox(
label="Opponent Steam player IDs",
placeholder="Optional, comma-separated for future player stat weighting",
max_lines=2,
)
with gr.Group():
gr.Markdown("### OpenDota API token", elem_classes=["section-title"])
opendota_token = gr.Textbox(
label="OpenDota API token",
type="password",
max_lines=1,
placeholder="Optional — leave blank to use default limits",
)
gr.Markdown(
"Detected from the environment — leave blank to use it."
if bool(os.getenv("OPENDOTA_API_KEY", ""))
else "Optional. Adds higher OpenDota rate limits if provided."
)
start_button = gr.Button("Start draft", variant="primary", size="lg")
with gr.Column(visible=False) as draft_page:
gr.HTML(
"""
"""
)
current_action = gr.Markdown(
"Initialize the draft to begin.",
elem_classes=["action-banner"],
)
with gr.Row(equal_height=False):
with gr.Column(scale=5):
with gr.Group():
with gr.Row():
hero_input = gr.Dropdown(
choices=HERO_SUGGESTIONS,
allow_custom_value=True,
filterable=True,
label="Hero",
value="",
scale=3,
)
submit_button = gr.Button(
"Submit step",
variant="primary",
interactive=False,
scale=1,
)
with gr.Row():
undo_button = gr.Button("Undo last step", interactive=False)
reset_button = gr.Button("Reset picks & bans")
board = gr.HTML(build_board(new_state()))
timeline = gr.Dataframe(
headers=["#", "Phase", "Action", "Team", "Hero", "Status"],
datatype=["str", "str", "str", "str", "str", "str"],
label="Draft path",
interactive=False,
wrap=True,
)
with gr.Column(scale=4):
gr.Markdown("### Recommendation", elem_classes=["section-title"])
insight_box = gr.Markdown(
"Backend updates will appear here as soon as you submit a pick or ban."
)
with gr.Accordion("Backend payload", open=False):
payload_box = gr.JSON()
outputs = [
state,
current_action,
hero_input,
board,
timeline,
insight_box,
payload_box,
submit_button,
undo_button,
]
start_button.click(
start_draft,
inputs=[
opendota_token,
user_team_name,
opponent_team_name,
user_side,
first_drafter,
opponent_athlete,
user_player_ids,
opponent_player_ids,
state,
],
outputs=outputs + [setup_page, draft_page],
)
submit_button.click(submit_hero, inputs=[hero_input, state], outputs=outputs)
undo_button.click(undo_last, inputs=state, outputs=outputs)
reset_button.click(reset_draft, inputs=state, outputs=outputs)
if __name__ == "__main__":
demo.launch(theme=THEME, css=CSS)