One Python process, about 250 lines, a mock server that runs with no API key, and the reconnect logic most tutorials skip.
Disclosure: Before you run this in a public server, read your data provider's terms on displaying and redistributing data.
What We're Building
A Discord bot that:
Holds a persistent WebSocket connection to a live sports feed.
Detects goals, score changes, red cards, half time and full time.
Posts a clean embed to a channel you choose.
Lets each server follow specific teams with slash commands (/follow, /unfollow, /following, /live).
Survives dropped connections with capped exponential backoff, and catches up on missed events via REST after every reconnect.
Ignores duplicate messages, so nobody gets double pings.
Final layout:
score-bot/
├── core.py # Update model, normalize(), Tracker (no Discord code, easy to test)
├── bot.py # Discord client, slash commands, stream loop, dispatcher
├── mock_server.py # fake live match over WebSocket, no API key needed
├── test_core.py
├── requirements.txt
└── .env
Why WebSockets Instead of Polling?
Polling a scores endpoint every few seconds burns your request quota and still shows events late. A WebSocket keeps one connection open and the server pushes updates the moment they exist. If you want the concept first, read MDN's WebSockets API guide.
For this build I use Orbistats as the example provider. It lists 13 sports (Football, Basketball, American Football, Cricket, Tennis, Baseball, Esports, Combat Sports, Volleyball, Handball, Ice Hockey, Golf and Horse Racing), and its docs describe a persistent WebSocket API on stream.orbistats.com plus webhooks and a live scores API. The site also markets sub-50ms feed latency. That is a vendor claim, so measure it on your own plan before you promise your community "instant" alerts.
The same architecture works with any provider that offers REST plus WebSocket. Only normalize() and the subscribe message change.
Step 1: Create the Discord Application
Open the Discord Developer Portal and click New Application.
Go to Bot, click Reset Token, and copy it. Treat it like a password.
You do not need any privileged intents. Slash commands and sending messages don't require the message-content intent.
Under OAuth2 → URL Generator, tick scopes bot and applications.commands, and permissions Send Messages and Embed Links. Open the generated URL to invite the bot to your test server.
The library docs are here: discord.py documentation.
Step 2: Project Setup
macOS / Linux
bash
mkdir score-bot && cd score-bot
python -m venv .venv && source .venv/bin/activate
Windows (PowerShell)
powershell
mkdir score-bot; cd score-bot
python -m venv .venv
.venv\Scripts\Activate.ps1
requirements.txt:
discord.py>=2.4
websockets>=14
aiohttp>=3.9
python-dotenv>=1.0
pytest>=8
bash
pip install -r requirements.txt
.env (and add .env to .gitignore, never commit keys):
DISCORD_TOKEN=your_discord_bot_token
ORBISTATS_API_KEY=your_orbistats_key
STREAM_URL=wss://stream.orbistats.com
SPORTS=football
MOCK=1
Set MOCK=1 to use the local fake feed, and MOCK=0 for the real one. Libraries used: websockets, aiohttp, and python-dotenv.
Step 3: The Core Logic (core.py)
This file has no Discord or network code, so it is fast to test. It contains three things: a data model, an adapter that converts provider JSON into that model, and a Tracker that decides whether an update deserves an alert.
python
core.py
from collections import deque
from dataclasses import dataclass
@dataclass
class Update:
uid: str # unique message id from the feed ("" for REST snapshots)
fixture_id: str
sport: str
home: str
away: str
home_score: int
away_score: int
minute: str
status: str # live | halftime | finished
kind: str # goal | red_card | status | score
detail: str
def normalize(raw: dict, sport_hint: str = "") -> "Update | None":
"""ADAPTER: provider JSON -> Update.
The raw shape below is an ASSUMPTION modeled on typical live-score feeds:
{"id": "...", "fixture": {...}, "event": {"kind": "...", "detail": "..."}}
Check the WebSocket docs and the sandbox, then edit ONLY this function.
"""
fx = raw.get("fixture", raw)
if "id" not in fx or "home" not in fx or "away" not in fx:
return None # heartbeat, ack, or something we don't care about
ev = raw.get("event") or {}
try:
return Update(
uid=str(raw.get("id", "")) if "fixture" in raw else "",
fixture_id=str(fx["id"]),
sport=fx.get("sport") or raw.get("sport") or sport_hint,
home=str(fx["home"]),
away=str(fx["away"]),
home_score=int(fx.get("home_score") or 0),
away_score=int(fx.get("away_score") or 0),
minute=str(fx.get("minute") or ""),
status=str(fx.get("status") or "live").lower(),
kind=str(ev.get("kind") or "score").lower(),
detail=str(ev.get("detail") or ""),
)
except (TypeError, ValueError):
return None # malformed message, skip it
class Tracker:
"""Remembers scores, ignores duplicates, and says whether to alert."""
def __init__(self, max_seen: int = 5000):
self.state = {} # fixture_id -> (home, away, status)
self._seen = set()
self._order = deque(maxlen=max_seen) # bounded, so memory can't grow forever
def _remember(self, uid: str):
if len(self._order) == self._order.maxlen:
self._seen.discard(self._order[0])
self._order.append(uid)
self._seen.add(uid)
def check(self, u: Update, baseline_only: bool = False) -> "str | None":
if u.uid:
if u.uid in self._seen:
return None # duplicate message
self._remember(u.uid)
prev = self.state.get(u.fixture_id)
self.state[u.fixture_id] = (u.home_score, u.away_score, u.status)
if prev is None:
if baseline_only:
return None # first sight in a snapshot: record, don't alert
prev = (0, 0, "")
ph, pa, pstatus = prev
if u.kind == "red_card":
return "RED CARD"
if u.home_score + u.away_score > ph + pa:
return "GOAL" if u.kind == "goal" else "SCORE UPDATE"
if u.status != pstatus:
if u.status == "halftime":
return "HALF TIME"
if u.status == "finished":
return "FULL TIME"
return None # includes score corrections (decreases)
Three design choices are worth explaining:
Dedupe by message id. Feeds can resend, and reconnects can replay. A bounded deque plus a set gives O(1) lookups without leaking memory.
baseline_only for REST snapshots. When the bot first starts, a snapshot of 30 live matches must not trigger 30 alerts. It should just record the current scores. On later reconnects, the same snapshot path detects goals you missed while offline.
Score decreases never alert. A downward change is usually a correction (VAR, data fix), and spamming "GOAL" for it is worse than staying silent.
Step 4: A Mock Server (mock_server.py)
This lets you develop and demo the whole bot without an API key or a live match. It plays a scripted game, sends the goals twice to test dedupe, then drops the connection to test reconnect:
python
mock_server.py
import asyncio, json
import websockets
SCRIPT = [ # (delay_s, status, home_score, away_score, minute, kind, detail)
(1, "live", 0, 0, "1", "status", "Kick-off"),
(3, "live", 1, 0, "14", "goal", "A. Rivera"),
(3, "live", 1, 1, "31", "goal", "J. Okafor"),
(3, "live", 1, 1, "38", "red_card", "M. Silva"),
(3, "halftime", 1, 1, "45", "status", "Half time"),
(3, "live", 2, 1, "63", "goal", "A. Rivera"),
(3, "finished", 2, 1, "90", "status", "Full time"),
]
seq = 0
async def handler(ws):
global seq
try:
print("client subscribed:", await ws.recv())
for delay, status, hs, as_, minute, kind, detail in SCRIPT:
await asyncio.sleep(delay)
seq += 1
msg = json.dumps({
"type": "fixture_update", "id": f"m-{seq}",
"fixture": {"id": "9001", "sport": "football",
"home": "Riverside FC", "away": "Harbor United",
"home_score": hs, "away_score": as_,
"minute": minute, "status": status},
"event": {"kind": kind, "detail": detail},
})
await ws.send(msg)
if kind == "goal":
await ws.send(msg) # duplicate on purpose
await asyncio.sleep(2)
await ws.close(code=1011) # simulate a server-side drop
except websockets.ConnectionClosed:
pass
async def main():
async with websockets.serve(handler, "localhost", 8765):
print("mock feed on ws://localhost:8765")
await asyncio.get_running_loop().create_future() # run forever
if name == "main":
asyncio.run(main())
Step 5: The Bot (bot.py)
5a. Config, database and the client
python
bot.py
import asyncio, json, logging, os, random, sqlite3
import aiohttp
import discord
import websockets
from discord import app_commands
from dotenv import load_dotenv
from core import Tracker, Update, normalize
load_dotenv()
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("scorebot")
MOCK = os.getenv("MOCK") == "1"
TOKEN = os.getenv("DISCORD_TOKEN", "")
API_KEY = os.getenv("ORBISTATS_API_KEY", "")
REST_BASE = "https://api.orbistats.com/v1"
STREAM_URL = "ws://localhost:8765" if MOCK else os.getenv("STREAM_URL", "wss://stream.orbistats.com")
SPORTS = [s.strip() for s in os.getenv("SPORTS", "football").split(",") if s.strip()]
HEADERS = {"Authorization": f"Bearer {API_KEY}"}
PLACEHOLDER: copy the real subscribe message from the WebSocket docs
SUBSCRIBE = {"action": "subscribe", "channel": "live-scores", "sports": SPORTS}
db = sqlite3.connect("bot.db")
db.executescript("""
CREATE TABLE IF NOT EXISTS settings(guild_id INTEGER PRIMARY KEY, channel_id INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS follows(guild_id INTEGER, term TEXT, PRIMARY KEY(guild_id, term));
""")
tracker = Tracker()
class ScoreBot(discord.Client):
def init(self):
super().init(intents=discord.Intents.default())
self.tree = app_commands.CommandTree(self)
self.queue: asyncio.Queue = asyncio.Queue()
self.session: aiohttp.ClientSession | None = None
async def setup_hook(self):
self.session = aiohttp.ClientSession()
await self.tree.sync()
self._tasks = [
asyncio.create_task(stream_loop(self)),
asyncio.create_task(dispatch_loop(self)),
]
async def close(self):
for t in getattr(self, "_tasks", []):
t.cancel()
if self.session:
await self.session.close()
await super().close()
bot = ScoreBot()
5b. The stream loop (reconnect, resync, and handling messages)
This is the reliability core. It reconnects forever with capped exponential backoff plus jitter. For why jitter matters, read AWS's Exponential Backoff and Jitter. After each (re)connect it subscribes, then runs a REST resync so goals scored while you were offline aren't lost:
python
async def fetch_live(sport: str) -> list:
"""REST snapshot. Path pattern assumed: /v1/{sport}/live. Confirm in the docs."""
async with bot.session.get(f"{REST_BASE}/{sport}/live", headers=HEADERS,
timeout=aiohttp.ClientTimeout(total=10)) as r:
r.raise_for_status()
data = await r.json()
items = data if isinstance(data, list) else data.get("data", [])
return [u for u in (normalize({"fixture": f}, sport) for f in items) if u]
async def resync():
if MOCK:
return # the mock feed has no REST side
for sport in SPORTS:
try:
for u in await fetch_live(sport):
title = tracker.check(u, baseline_only=True)
if title:
await bot.queue.put((title, u)) # catch-up alert for a missed event
except Exception:
log.exception("resync failed for %s", sport)
async def stream_loop(b: ScoreBot):
await b.wait_until_ready()
backoff = 1
while True:
try:
# websockets >= 14 uses additional_headers. Older versions used extra_headers.
async with websockets.connect(STREAM_URL, additional_headers=HEADERS,
ping_interval=20, ping_timeout=20) as ws:
log.info("stream connected: %s", STREAM_URL)
backoff = 1
await ws.send(json.dumps(SUBSCRIBE))
await resync()
async for raw in ws:
try:
u = normalize(json.loads(raw))
except json.JSONDecodeError:
continue
if not u:
continue
title = tracker.check(u)
if title:
await b.queue.put((title, u))
except Exception as e:
log.warning("stream dropped (%s). retrying in ~%ss", e, backoff)
await asyncio.sleep(backoff + random.random())
backoff = min(backoff * 2, 60)
Notes on the choices:
except Exception does not swallow task cancellation, so bot.close() still stops the loop cleanly.
ping_interval makes the library detect half-dead connections. If the server closes with a code, the close-code list on MDN explains what each one means.
The auth method here is a Bearer header. Your provider may require a token in the URL or in the first message instead, so follow its docs.
If your provider offers only webhooks instead of a socket, the dispatcher below stays identical. Only the intake changes.
5c. The dispatcher (Discord side)
python
EMOJI = {"GOAL": "⚽", "SCORE UPDATE": "📣", "RED CARD": "🟥", "HALF TIME": "⏸️", "FULL TIME": "🏁"}
def esc(s: str) -> str:
return discord.utils.escape_mentions(discord.utils.escape_markdown(s))
def make_embed(title: str, u: Update) -> discord.Embed:
e = discord.Embed(
title=f"{EMOJI.get(title, '📣')} {title}",
description=f"{esc(u.home)} {u.home_score} – {u.away_score} {esc(u.away)}",
colour=discord.Colour.green() if title in ("GOAL", "SCORE UPDATE") else discord.Colour.orange(),
)
if u.detail:
e.add_field(name="Details", value=esc(u.detail)[:200], inline=True)
e.set_footer(text=f"{u.sport} · {u.minute}' · {u.status}")
return e
def wants(guild_id: int, u: Update) -> bool:
terms = [r[0] for r in db.execute("SELECT term FROM follows WHERE guild_id=?", (guild_id,))]
hay = f"{u.home} {u.away} {u.sport}".lower()
return any(t == "*" or t in hay for t in terms)
async def dispatch_loop(b: ScoreBot):
await b.wait_until_ready()
while True:
title, u = await b.queue.get()
try:
for guild_id, channel_id in db.execute("SELECT guild_id, channel_id FROM settings").fetchall():
if not wants(guild_id, u):
continue
channel = b.get_channel(channel_id)
if channel is None:
continue
try:
await channel.send(embed=make_embed(title, u),
allowed_mentions=discord.AllowedMentions.none())
except discord.HTTPException as e:
log.warning("send failed in %s: %s", channel_id, e)
await asyncio.sleep(0.3) # gentle pacing, discord.py also handles 429s
except Exception:
log.exception("dispatch error") # one bad event must not kill the loop
Two safety details matter here. Team names and player names come from an external feed, so esc() strips markdown and mentions, and AllowedMentions.none() guarantees the bot can never ping @everyone because of a weird string in the data.
5d. Slash commands
python
@bot.tree.command(description="Post live alerts in this channel")
@app_commands.guild_only()
@app_commands.default_permissions(manage_guild=True)
async def setchannel(interaction: discord.Interaction):
db.execute("INSERT INTO settings(guild_id, channel_id) VALUES(?,?) "
"ON CONFLICT(guild_id) DO UPDATE SET channel_id=excluded.channel_id",
(interaction.guild_id, interaction.channel_id))
db.commit()
await interaction.response.send_message("Alerts will post in this channel.", ephemeral=True)
@bot.tree.command(description="Follow a team or sport (use * for everything)")
@app_commands.guild_only()
@app_commands.default_permissions(manage_guild=True)
async def follow(interaction: discord.Interaction, term: str):
term = term.strip().lower()[:50]
db.execute("INSERT OR IGNORE INTO follows(guild_id, term) VALUES(?,?)", (interaction.guild_id, term))
db.commit()
await interaction.response.send_message(f"Following {term}.", ephemeral=True)
@bot.tree.command(description="Stop following a term")
@app_commands.guild_only()
@app_commands.default_permissions(manage_guild=True)
async def unfollow(interaction: discord.Interaction, term: str):
db.execute("DELETE FROM follows WHERE guild_id=? AND term=?", (interaction.guild_id, term.strip().lower()))
db.commit()
await interaction.response.send_message(f"Unfollowed {term}.", ephemeral=True)
@bot.tree.command(description="List what this server follows")
@app_commands.guild_only()
async def following(interaction: discord.Interaction):
terms = [r[0] for r in db.execute("SELECT term FROM follows WHERE guild_id=?", (interaction.guild_id,))]
await interaction.response.send_message(", ".join(f"{t}" for t in terms) or "Nothing yet. Try /follow *.",
ephemeral=True)
@bot.tree.command(description="Show current live matches")
async def live(interaction: discord.Interaction, sport: str = "football"):
if MOCK:
await interaction.response.send_message("REST snapshots are disabled in mock mode.", ephemeral=True)
return
await interaction.response.defer(ephemeral=True) # REST call may exceed Discord's 3s window
try:
items = (await fetch_live(sport.strip().lower()))[:10]
except Exception:
await interaction.followup.send("Couldn't fetch live scores right now.", ephemeral=True)
return
lines = [f"{esc(u.home)} {u.home_score}–{u.away_score} {esc(u.away)} ({u.minute}')" for u in items]
await interaction.followup.send("\n".join(lines) or "No live matches.", ephemeral=True)
if name == "main":
if not TOKEN:
raise SystemExit("Set DISCORD_TOKEN in .env")
bot.run(TOKEN)
/live uses defer() on purpose, because Discord requires an initial response within 3 seconds and a network call can exceed that.
Step 6: Run It
Terminal 1:
bash
python mock_server.py
Terminal 2:
bash
python bot.py
Then in your test server:
/setchannel in the channel where alerts should appear.
/follow * (or /follow riverside).
Watch the mock match play out. You should get alerts for the goal at 14', the goal at 31', a red card, half time, the goal at 63', and full time. The duplicated goal messages should not produce duplicate embeds.
About two seconds after full time the mock server drops the connection. Check your logs for stream dropped ... retrying, followed by a reconnect and a new replay of the script.
Slash commands may take a moment to appear the first time. If they don't, restart your Discord client.
Step 7: Point It at the Real Feed
Get a key (see the pricing page for current free-tier limits) and open the documentation and the sandbox.
Copy a real WebSocket message and edit only normalize() so it produces an Update. Nested objects such as home: {"name": ...} are common, so adapt the field access.
Copy the real subscribe payload into SUBSCRIBE, and confirm how the key is sent (header, query string, or first message).
Set MOCK=0 in .env, and confirm the REST snapshot path and response shape used by fetch_live().
Read the changelog before shipping, since APIs change.
Other sports work the same way: the provider describes one API shape across its sports, but score semantics differ. Cricket has runs and wickets, and tennis has sets and games, so extend Tracker.check() per sport instead of assuming "score went up" means "goal."
Step 8: Tests (test_core.py)
python
test_core.py
from core import Tracker, Update, normalize
def mk(uid, h, a, status="live", kind="score", fid="1"):
return Update(uid, fid, "football", "A", "B", h, a, "10", status, kind, "")
def test_goal_alert():
assert Tracker().check(mk("1", 1, 0, kind="goal")) == "GOAL"
def test_duplicate_ignored():
t = Tracker()
assert t.check(mk("1", 1, 0, kind="goal")) == "GOAL"
assert t.check(mk("1", 1, 0, kind="goal")) is None
def test_snapshot_baseline_does_not_alert():
assert Tracker().check(mk("", 1, 0), baseline_only=True) is None
def test_catchup_after_reconnect():
t = Tracker()
t.check(mk("", 1, 0), baseline_only=True)
assert t.check(mk("", 2, 0), baseline_only=True) == "SCORE UPDATE"
def test_score_correction_is_silent():
t = Tracker()
t.check(mk("1", 2, 0, kind="goal"))
assert t.check(mk("2", 1, 0)) is None
def test_red_card_and_full_time():
t = Tracker()
assert t.check(mk("1", 0, 0, kind="red_card")) == "RED CARD"
assert t.check(mk("2", 0, 0, status="finished")) == "FULL TIME"
def test_normalize_ignores_junk():
assert normalize({"type": "heartbeat"}) is None
bash
pytest -q
Production Checklist
Secrets: token and API key live in environment variables or a secrets manager, never in the repo and never in a client. Keep the WebSocket key server-side.
Resync after every reconnect, already built in. Otherwise you silently miss events during outages.
Dedupe and ordering: dedupe is built in. If your provider sends sequence numbers, also drop any message older than the last one you processed per fixture.
Backoff with jitter, already built in, so a provider outage doesn't turn into a thundering herd.
Escape everything from the feed before it reaches Discord (done with esc() and AllowedMentions.none()).
Persistence: SQLite is fine for one bot. For multiple instances, move follows into a shared database, and make sure only one process holds the stream connection.
Monitoring: log every reconnect, and alert yourself if the stream has been down for more than a few minutes. If the feed is quiet outside match hours, don't treat silence alone as a failure.
Terms of use: confirm you're allowed to display the data publicly, and add attribution if required.
Measure: compare an alert's timestamp to a trusted broadcast during a real match, because network latency and data delay are different numbers.
Deploying (Optional)
A minimal Dockerfile:
dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY core.py bot.py ./
CMD ["python", "bot.py"]
bash
docker build -t score-bot .
docker run -d --restart unless-stopped --env-file .env -v $(pwd)/data:/app score-bot
Mount a volume for bot.db so follows survive restarts. See the Docker documentation for the full details.
Ideas for Part 2
Per-match threads: create a Discord thread per fixture and stream every event into it.
Reaction roles: let members pick teams by clicking buttons instead of typing /follow.
More sports: add cricket wickets, tennis set wins, and basketball run alerts (for example, an 8-0 run).
Webhook intake: swap the socket for webhooks if you'd rather host an HTTPS endpoint than hold a connection.
Web dashboard: if you'd rather see a scoreboard in the browser, Orbistats has a 50-line JavaScript live scoreboard tutorial, and a Python + FastAPI tutorial shows caching for the REST side.
Recap
You built a bot that holds a live connection, turns raw feed messages into clean alerts, ignores duplicates and corrections, heals itself after disconnects, and lets each Discord server choose what it follows. The real work isn't the Discord part. It is the boring reliability layer: backoff, resync, dedupe, and escaping. Get those right and adding new sports is mostly an afternoon of mapping fields.
Which sport should I add in Part 2? Tell me in the comments.
Top comments (0)