---
title: "Telegram Music Bot Tutorial — aiogram 3 + file_id | FastSaverAPI"
url: https://fastsaverapi.com/guides/telegram-music-bot/
description: "Build a Telegram music bot in aiogram 3: search YouTube Music, answer each tap with a cached file_id, catch every error path. Full code and the credit maths."
updated: 2026-08-10
api_base: https://api.fastsaver.io/v1
site_index: https://fastsaverapi.com/llms.txt
openapi: https://fastsaverapi.com/openapi.json
---

# Telegram music bot that never downloads a file

A Telegram music bot in three files: the user types a song name, your bot searches YouTube
Music, and a tap sends the track as a Telegram file_id — no download, no upload, no
ffmpeg. Full aiogram 3 code, the error paths included, and an honest count of what each song
costs in credits .

- **Stack**: Python 3.11+ · aiogram 3 · httpx

- **Endpoints**: GET /youtube/search · POST /youtube/audio/tg-bot

- **Cost**: 2 to search · 7 per track cached, 15 on a miss

- **Bytes through your server**: none

## what you are building

The bot has two states. A user sends text, you search YouTube Music and reply with an inline
keyboard of matches. The user taps a row, you turn that row's video ID into a Telegram
file_id and send it as audio.

Note what is missing: no MP3 on disk, no temp directory, no multipart upload leaving your
process. Your bot passes a string where a file would go and Telegram serves the bytes. Two API
calls per song — nine credits when we already hold the track, seventeen the first time anyone asks
for it — and three files: a client, a cache, the handlers.

## setup

You need a bot token from BotFather and a FastSaverAPI key. Keep both in the environment; the
token alone lets anyone impersonate your bot.



```
python -m venv .venv && source .venv/bin/activate
pip install "aiogram>=3.7" httpx

export BOT_TOKEN="123456789:AA..."          # from @BotFather
export FASTSAVER_KEY="fs_sk_•••••••••••"    # from your dashboard
```

Pin aiogram 3 deliberately: snippets written for 2.x will not run here, and from 3.7 defaults
such as parse_mode moved onto DefaultBotProperties . Pick the bot username
now and treat it as configuration — a file_id is issued for one bot and rejected by
every other. The Telegram bot media API page explains what
a file_id actually is and how the scoping rule works; this guide just obeys it.

## step one — find the track

GET /youtube/search queries YouTube Music rather than the main site, so "tame
impala" returns tracks instead of reaction videos. Both query and page
are required.

GET
https://api.fastsaver.io/v1/youtube/search
2 credits · page 1–3



```
curl "https://api.fastsaver.io/v1/youtube/search?query=tame%20impala&page=1" \
  -H "X-Api-Key: $FASTSAVER_KEY"
```



```
{
  "ok": true,
  "page": 1,
  "results": [
    {
      "video_id": "5K2ADZ7gzMc",
      "title": "Tame Impala - The Less I Know The Better",
      "duration": "3:38",
      "thumbnail": "https://i.ytimg.com/vi/5K2ADZ7gzMc/mqdefault.jpg",
      "thumbnail_max": "https://i.ytimg.com/vi/5K2ADZ7gzMc/maxresdefault.jpg"
    }
  ]
}
```


Two details matter downstream. duration is a display string, not seconds. And
video_id is the only field the next call needs; the rest exists so you can build a
readable keyboard without a second lookup. The full field reference — and the other three YouTube
endpoints — lives on the YouTube downloader API page.

## step two — the API client

Wrap both endpoints once, in a class that turns HTTP failures into exceptions your handlers can
branch on — and treats ok , not the status code, as the success flag.



```
# fastsaver.py
import httpx

# Running out of credits is a plain HTTP 400 — there is no 402 anywhere in this
# API — and 400 is also what an unresolvable link returns, so the detail string
# is the only thing that tells the two apart.
INSUFFICIENT = "Insufficient credits"


class ApiError(RuntimeError):
    """The API refused the request."""


class OutOfCredits(ApiError):
    pass


class RateLimited(ApiError):
    pass


class FastSaver:
    def __init__(self, key: str, bot_username: str) -> None:
        self.bot_username = bot_username
        self._http = httpx.AsyncClient(
            base_url="https://api.fastsaver.io/v1",
            headers={"X-Api-Key": key},
            timeout=httpx.Timeout(120.0, connect=10.0),
        )

    async def _call(self, method: str, path: str, **kw) -> dict:
        r = await self._http.request(method, path, **kw)
        if r.status_code == 401:
            raise ApiError("api key missing or wrong")
        if r.status_code == 429:
            raise RateLimited("plan rate limit reached")
        data = r.json()
        if not data.get("ok"):
            detail = data.get("detail") or f"http {r.status_code}"
            if INSUFFICIENT in detail:      # 400, not 402
                raise OutOfCredits(detail)
            raise ApiError(detail)
        return data

    async def search(self, query: str, page: int = 1) -> list[dict]:
        data = await self._call(
            "GET", "/youtube/search", params={"query": query, "page": page},
        )
        return data["results"]

    async def audio_file_id(self, video_id: str) -> str:
        data = await self._call(
            "POST", "/youtube/audio/tg-bot",
            json={"video_id": video_id, "bot_username": self.bot_username},
        )
        return data["file_id"]

    async def balance(self) -> dict:
        return await self._call("GET", "/balance")
```

The status codes are worth reading twice, because the obvious guess is wrong. An exhausted
balance does not come back as 402 Payment Required ; it comes back as
400 with
{"ok": false, "detail": "Insufficient credits. Please top up to your account."} .
Branch on 402 and the branch is simply never taken — your users get whatever generic
message you kept for the unknown case, and you find out you are broke from a support message
rather than a log line. Match on the detail text instead. 401 and 429
really are their own statuses, so those two stay where they are.

The 120-second read timeout is not padding: most tracks answer immediately from our cache, but
one nobody has asked for before is fetched on demand, and that is measured in seconds. It is also
billed as a full download — 15 credits rather than 7 — so a miss costs you on both axes. The
connect timeout stays short so a dead network fails fast.

## step three — the handlers

Two handlers: one for text, one for taps. callback_data is capped at 64 bytes, so
only the eleven-character video ID travels in it; titles live in a dict keyed by the same ID.



```
# bot.py
import asyncio, logging, os
from aiogram import Bot, Dispatcher, F
from aiogram.exceptions import TelegramBadRequest
from aiogram.types import CallbackQuery, Message
from aiogram.utils.keyboard import InlineKeyboardBuilder

from cache import drop_file_id, get_file_id, put_file_id
from fastsaver import ApiError, FastSaver, OutOfCredits, RateLimited

BOT_USERNAME = "@example_music_bot"

bot = Bot(os.environ["BOT_TOKEN"])
dp = Dispatcher()
api = FastSaver(os.environ["FASTSAVER_KEY"], BOT_USERNAME)
titles: dict[str, str] = {}


def excuse(exc: ApiError) -> str:
    if isinstance(exc, OutOfCredits):
        return "The bot is out of credits. Ping the owner."
    if isinstance(exc, RateLimited):
        return "Busy right now — try again in a minute."
    return "Something broke on our side. Try again."


@dp.message(F.text & ~F.text.startswith("/"))
async def on_text(msg: Message) -> None:
    query = msg.text.strip()[:100]
    try:
        results = await api.search(query)
    except ApiError as exc:
        logging.warning("search failed for %r: %s", query, exc)
        await msg.answer(excuse(exc))
        return

    if not results:
        await msg.answer("No tracks matched that. Try the artist name too.")
        return

    kb = InlineKeyboardBuilder()
    for track in results[:8]:
        titles[track["video_id"]] = track["title"]
        kb.button(
            text=f"{track['title']}  ·  {track['duration']}",
            callback_data=f"a:{track['video_id']}",
        )
    kb.adjust(1)
    await msg.answer("Pick a track:", reply_markup=kb.as_markup())


@dp.callback_query(F.data.startswith("a:"))
async def on_pick(cb: CallbackQuery) -> None:
    await cb.answer("Fetching…")       # ack now, or the button spins forever
    video_id = cb.data[2:]

    file_id = get_file_id(video_id, BOT_USERNAME)
    if file_id is None:
        try:
            file_id = await api.audio_file_id(video_id)
        except ApiError as exc:
            logging.warning("resolve failed for %s: %s", video_id, exc)
            await cb.message.answer(excuse(exc))
            return
        put_file_id(video_id, BOT_USERNAME, file_id)

    try:
        await cb.message.answer_audio(file_id, title=titles.get(video_id))
    except TelegramBadRequest:
        drop_file_id(video_id, BOT_USERNAME)
        await cb.message.answer("That reference went stale. Tap it again.")


async def main() -> None:
    logging.basicConfig(level=logging.INFO)
    acct = await api.balance()         # free call, never spends credits
    logging.info("plan=%s credits=%s rpm=%s",
                 acct["plan"], acct["credits"], acct["rpm_limit"])
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())
```

The first line of on_pick is the one people leave out. Telegram wants a prompt
answer to a callback query; resolve the track first, let a cold fetch take six seconds, and the
user watches a spinner and taps again — you have now paid for one song twice, 15 credits and then
7.

excuse() only has an out-of-credits message to show because the client bothered to
read the detail string. Classify in one place, phrase in another: handlers should ask
"which exception is this", never "which status code was that".

The startup call to /balance is free and puts your plan, credits and rate limit in
the first log line — the cheapest way to learn that this container has the wrong key.

## step four — remember the file_id

The delivery endpoint costs 7 credits when we already hold the track and 15 when we have to
fetch it first. Sending a file_id you already hold costs nothing at all. So store
them, keyed by video ID and bot username.



```
# cache.py
import sqlite3

db = sqlite3.connect("tracks.db", check_same_thread=False)
db.execute("""
    CREATE TABLE IF NOT EXISTS file_ids (
        video_id TEXT NOT NULL,
        bot      TEXT NOT NULL,
        file_id  TEXT NOT NULL,
        PRIMARY KEY (video_id, bot)
    )
""")
db.commit()


def get_file_id(video_id: str, bot: str) -> str | None:
    row = db.execute(
        "SELECT file_id FROM file_ids WHERE video_id = ? AND bot = ?",
        (video_id, bot),
    ).fetchone()
    return row[0] if row else None


def put_file_id(video_id: str, bot: str, file_id: str) -> None:
    db.execute("INSERT OR REPLACE INTO file_ids VALUES (?, ?, ?)",
               (video_id, bot, file_id))
    db.commit()


def drop_file_id(video_id: str, bot: str) -> None:
    db.execute("DELETE FROM file_ids WHERE video_id = ? AND bot = ?",
               (video_id, bot))
    db.commit()
```

Yes, sqlite3 blocks the event loop. At a few hundred taps an hour the queries are
microseconds; when that stops being true, move the table behind an async driver and change three
functions. Do not start there.

The composite key is not decoration. Migrate to a second bot and every stored string is useless
to it — with the bot in the key it just misses the cache and re-resolves.

## the paths that are not the happy one

Seven routine failures. Catch every one at the handler — a single bad video ID should never stop
the polling loop. Two of them share HTTP 400, so read the detail , not the status.

| Failure | What to do |
| --- | --- |
| Empty results | Nothing matched and the 2 credits are spent. Ask for the artist name; the same text will fail again. |
| 400 · Insufficient credits. Please top up to your account. | Out of credits — this exact detail string, on a 400. There is no 402 to catch. Nothing works until you top up, so watch /balance from a health check; that call is free. |
| 400 · any other detail | The track is private, region-locked, age-gated or gone. A failed resolve is billed at a flat 0.1 credits, so tell the user and move on — a retry loop buys the same answer at 0.1 a go. |
| 429 | Past your plan's per-minute limit: 10 Free, 60 Pro, 900 Mega. Back off, then queue the resolve calls. |
| 401 | The key never reached us — usually an unset environment variable. |
| TelegramBadRequest | A stale file_id. Drop the row, let the next tap re-resolve. |
| TelegramRetryAfter | Telegram's own flood control. Wait the seconds it gives you. |

That first 400 is the one worth wiring into your alerting. It is the only failure on the list
that will not fix itself, and because it arrives with the same status as a dead link, a client that
only looks at r.status_code reports "that track is unavailable" for every song in the
catalogue.

## what it actually costs

Three prices, not two: 2 credits to search, then 7 to turn a video ID into a
file_id if we already hold that track — and 15 if we do not, because the first
request for a track pays for the download. Resending something already in your own table is free.

| Interaction | API calls | Credits |
| --- | --- | --- |
| Search, tap a track we already hold | 2 | 9 |
| Search, tap a track nobody has pulled yet | 2 | 17 |
| Delivery only — id from charts or a deep link | 1 | 7 or 15 |
| A track already in your cache table | 0 | 0 |
| A resolve that fails | 1 | 0.1 |

You cannot tell which price a tap will pay until it has been paid, and our cache is shared, so
the odds improve as more people ask for the same music. Plan with a miss rate rather than a single
number: one tap in five landing on something new puts the average flow at 10.6 credits.

The free tier's 1,000 credits therefore buy 111 search-and-send flows if every
track is already cached, 58 if none are, and about 94 at that
one-in-five rate. Deliveries on their own, when the video ID comes from a chart or a deep link:
142 cached, 66 uncached. Enough to test properly; not enough to launch.

The $9 Pro plan's 100,000 credits a month buy roughly 11,100 flows all-cached,
9,400 at a one-in-five miss rate, and 5,880 in the pathological
case where every track is new — 81 cents, 95 cents and $1.53 per thousand songs. Credits bind long
before the rate limit does: 9,400 flows a month averages thirteen an hour, nowhere near Pro's 60
requests a minute.

That spread is the argument for the table in step four. Your own file_id cache does
not just shave 7 credits off a repeat send — it takes the second and every later delivery of a
track to zero, which is what pulls a real bot's average toward the good end of that range. Music
requests are heavily repetitive; a few hundred rows of SQLite absorb most of them.

## where a Telegram music bot hits its ceiling

- Public content only. If a track needs a login, a membership or a different country to play, the ID will not resolve.
- A file_id belongs to one bot. Not portable to a second bot, not to a user account.
- A cache miss costs more, not just longer. 15 credits instead of 7, on a delay you cannot predict — budget for both.
- Two rate limits, two owners. Ours is per plan, per minute. Telegram polices your bot separately.
- YouTube changes; your bot does not. Breakage is repaired behind the endpoint, so a bot deployed months ago keeps answering without a redeploy.

And the part no library handles: a bot with an audience distributing recorded music is publishing
it. Whether you may is between you, the rights holders and Telegram's terms.

## questions about building the bot

**Q: Does the bot need yt-dlp or ffmpeg installed?**

A: No. Nothing is decoded or converted on your machine — aiogram and an HTTP
client are the whole dependency list.

**Q: How many songs can the bot deliver on the free plan?**

A: Between 58 and 111, and where you land depends on our cache. A flow is 2 credits for
/youtube/search plus 7 for /youtube/audio/tg-bot when we already hold
the track, or 15 when we have to fetch it first — 9 or 17 credits against the 1,000 free ones.
Budget for about 94 if you assume one tap in five is a track nobody has pulled before. Skip the
search and it is 142 cached, 66 uncached.

**Q: Which status code means the bot has run out of credits?**

A: HTTP 400 , with detail set to "Insufficient credits. Please top
up to your account." There is no 402 in this API, so a client that branches on 402 silently
never sees the case. Match the detail string, and remember 400 also covers a link that cannot be
resolved — the status alone does not tell you which happened.

**Q: Why does the button keep spinning after a user taps a track?**

A: The callback query was not answered in time. Resolving an uncached track takes seconds, so
call cb.answer() on the first line of the handler and send the audio afterwards as
a normal message.

**Q: Can I use inline mode instead of an inline keyboard?**

A: Yes — InlineQueryResultCachedAudio takes a file_id . The catch is
arithmetic: you need one per result shown, so ten results for a single keystroke is 70 credits
if every track is cached on our side and 150 if none of them are. Serve inline results from
your cache table only.

**Q: How many search results can one query return?**

A: Ten per page, and page accepts 1, 2 or 3 — thirty at most, 2 credits per page.
A keyboard longer than eight rows is unreadable on a phone, so page 1 is usually the whole
feature.

## Keep reading

- Telegram bot media API — the endpoint reference
- YouTube downloader API — search, info, download
- Shazam API — identify a forwarded voice note
- YouTube to MP3 over HTTP — when you need the file

## Build it this evening

1,000 free credits is 58 to 111 songs, depending on the cache — more than enough to get the bot working end to end.

Get an API key
Try the endpoints
